Cara membangun saluran pipa ETL dengan Python

Diterbitkan: 2022-01-11

ETL adalah singkatan dari E xtract, T ransform, L oad. Sebagai bagian dari proses ETL, data Diekstraksi, Ditransformasi, dan Dimuat ke Gudang Data sehingga organisasi dapat menganalisisnya untuk membuat keputusan strategis.

Berikut ini adalah langkah-langkah utama yang dilakukan dalam pipa ETL:

Ekstrak: Proses ini mengumpulkan dan mengintegrasikan data dari berbagai sumber, termasuk Database, Data Lakes, CRM, dan lainnya.
Transform: Ini adalah fase paling penting dalam ETL Pipeline. Untuk membuat data, siap analitik, data harus dikumpulkan, diurutkan, dibersihkan, dan diputar dengan benar, dalam langkah ini.
Muat: Proses ini melibatkan pengimporan data terstruktur atau tidak terstruktur dari Data Lakes, Database, dan sumber lain ke dalam Gudang Data sehingga Analis Data atau pengguna lain dapat memperoleh wawasan mendalam dengan mudah.

Memahami Pentingnya Python ETL

Python adalah salah satu bahasa pemrograman paling populer & paling sering digunakan di dunia modern, dengan aplikasi tanpa akhir di berbagai bidang. Itu telah memenangkan penghargaan TIOBE Programming Language of the Year 2021 yang bergengsi.

Sifat Python yang fleksibel dan dinamis membuatnya ideal untuk tugas Deployment, Analisis, dan Pemeliharaan. Python ETL adalah salah satu keterampilan penting yang diperlukan dalam Rekayasa Data untuk membangun Pipa Data, mengembangkan Model Statistik, dan melakukan analisis menyeluruh terhadapnya.

Ini telah menjadi alat yang populer untuk mengeksekusi proses ETL karena kemudahan penggunaan dan perpustakaan yang kuat untuk mengakses database dan sistem penyimpanan. Banyak tim menggunakan Python untuk ETL & Rekayasa Data daripada alat ETL karena lebih fleksibel dan kuat untuk tugas-tugas ini.

Manfaat terbesar Python dibandingkan bahasa pemrograman lain adalah kesederhanaan penggunaan dalam Penambangan Data, Ilmu Data, Data Besar, Kecerdasan Buatan, dan Pembelajaran Mesin.

Perusahaan di seluruh dunia menggunakan Python untuk data mereka guna memperoleh wawasan, mengelola operasi mereka, dan menjaga semuanya berjalan lancar.

2 Langkah Mudah untuk Membangun Python ETL Pipeline

Di bagian ini, Anda akan mempelajari langkah-langkah penting untuk membangun pipeline ETL menggunakan Python . Anda akan membuat Data Pipeline dasar yang memasukkan data ke dalam database Microsoft SQL Server dari Database MySQL & Microsoft SQL Server.

Untuk menyiapkan skrip Python ETL, ikuti langkah-langkah di bawah ini:

Langkah 1: Instal Modul yang Diperlukan

Untuk menyiapkan Python ETL Pipeline, Anda harus menginstal modul berikut:

Konektor Python ke MySQL: mysql-connector-python (Gunakan perintah pip install mysql-connector-python untuk menginstal)
Python ke Microsoft SQL Server Connector: pyodbc (Gunakan perintah pip install pyodbc untuk menginstal)

Langkah 2: Siapkan Direktori ETL

Setelah menginstal paket-paket di atas, Anda perlu membuat 4 file Python, yang disebutkan di bawah ini di direktori proyek Anda:

db_credentials.py: File ini menyertakan kode untuk membuat koneksi dengan semua Database.
sql_queries.py: File ini terdiri dari kueri Database yang umum digunakan untuk mengekstrak dan memuat data dalam format string.
etl.py: File ini memiliki operasi yang diperlukan untuk terhubung ke Database dan menjalankan kueri yang diperlukan.
main.py: Ini adalah file utama yang mengatur aliran dan eksekusi dari Python ETL Pipeline.

A) db_credentials.py

Semua String Koneksi Database Sumber dan Target harus disertakan dalam file ini. Ini harus berisi semua informasi yang diperlukan untuk mengakses database yang relevan dalam format daftar sehingga dapat dengan cepat diulang bila diperlukan. Berikut ini adalah contoh skrip Python untuk membuat koneksi Database:

 datawarehouse_name = 'nama_dwh_anda'
# sql-server (target db, gudang data)
datawarehouse_db_config = {
  'Koneksi_Tepercaya': 'ya',
  'pengandar': '{SQL Server}',
  'server': 'datawarehouse_sql_server',
  'database': '{}'.format(nama_warehouse),
  'pengguna': 'nama_db_Anda',
  'kata sandi': 'kata_db_anda',
  'autocommit': Benar,
}
# sumber db > sql-server
sqlserver_db_config = [
  {
    'Koneksi_Tepercaya': 'ya',
    'pengandar': '{SQL Server}',
    'server': 'server_db_sql_Anda',
    'database': 'db_1',
    'pengguna': 'nama_db_Anda',
    'kata sandi': 'kata_db_anda',
    'autocommit': Benar,
  }
]
# sumber db > mysql
mysql_db_config = [
  {
    'pengguna': 'pengguna_1_Anda',
    'sandi': 'kata__anda',
    'host': 'db_connection_string_1',
    'database': 'db_1',
  },
  {
    'pengguna': 'pengguna_2_anda,
    'sandi': 'sandi_2_anda',
    'host': 'db_connection_string_2',
    'database': 'db_2nd',
  },
]

B) sql_queries.py

File ini mencakup kueri untuk mengekstraksi data dari Database Sumber dan memuatnya ke dalam Database Target. Skrip berikut akan membantu Anda melakukan tugas ini:

 # contoh kueri, akan unik untuk platform basis data yang berbeda

sqlserver_extract = ('''
  PILIH sqlserver_col_1, sqlserver_col_2, sqlserver_col_3
  DARI sqlserver_1_table
''')
sqlserver_insert = ('''
  INSERT INTO table_demo (col_1, col_2, col_3)
  NILAI (?, ?, ?)  
''')
mysql_extract = ('''
  PILIH mysql_col_1, mysql_col_2, mysql_col_3
  DARI mysql_demo_table
''')
mysql_insert = ('''
  INSERT INTO table_demo (col_1, col_2, col_3)
  NILAI (?, ?, ?)  
''')

# Kueri diekspor
kelas Sql_Query:
  def __init__(sendiri, ekstrak_query, load_query):
    self.extract_query = ekstrak_query
    self.load_query = load_query   
# buat instance untuk kelas Sql_Query
sqlserver_query = SqlQuery(sqlserver_extract, sqlserver_insert)
mysql_query = SqlQuery(mysql_extract, mysql_insert)
# membuat daftar untuk iterasi melalui nilai
mysql_queries = [mysql_query]
sqlserver_queries = [sqlserver_query]

C) etl.py

File ini harus menyertakan kode yang diperlukan untuk mengakses Database yang relevan dan menjalankan kueri yang diperlukan. Skrip berikut akan membantu Anda melakukan tugas ini:

 # modul berbasis python
impor pyodbc
impor mysql.connector

def etl(kueri, source_cnx, target_cnx):
  # ekstrak data dari basis data sumber demo
  sumber_kursor = sumber_cnx.kursor()
  source_cursor.execute(query.extract_query)
  data = source_cursor.fetchall()
  sumber_kursor.close()

  # memuat data ke demo Data Warehouse db
  
jika data:
    target_kursor = target_cnx.kursor()
    target_cursor.execute("GUNAKAN {}".format(name_for_datawarehouse))
    target_cursor.executemany(query.load_query, data)
    print('data dimuat ke demo Data Warehouse db')
    target_kursor.close()
  lain:
    print('data kosong')

def etl_process(permintaan, target_cnx, source_db_config, db_platform):

  # mengonfigurasi koneksi basis data sumber demo
  jika db_platform == 'mysql':
    source_cnx = mysql.connector.connect(**source_db_config)
  elif db_platform == 'sqlserver':
    source_cnx = pyodbc.connect(**source_db_config)
  lain:
    kembali 'Kesalahan! platform basis data sumber tidak dikenal'
  # loop melalui kueri sql
  untuk kueri dalam kueri:
    etl (kueri, source_cnx, target_cnx)    
  # tutup koneksi sumber db
  sumber_cnx.close()

D) main.py

File ini menyertakan kode untuk beralih melalui kredensial yang diberikan untuk terhubung ke database dan menjalankan operasi ETL Python yang diperlukan. Skrip berikut akan membantu Anda melakukan tugas ini:

 # variabel
dari db_credentials impor datawarehouse_db_config, sqlserver_db_config, mysql_db_config
dari sql_queries impor sqlserver_queries, mysql_queries

# metode
dari etl impor etl_process
def utama():
  print('memulai proses data etl')
	
  # buat koneksi untuk SQL Server, penyimpanan tujuan yang diinginkan
  target_cnx = pyodbc.connect(**datawarehouse_db_config)
	
  # mengulang melalui kredensial
  # Basis Data > mysql
  untuk konfigurasi di mysql_db_config: 
    mencoba:
      print("memuat db: " + config['database'])
      etl_process(mysql_queries, target_cnx, config, 'mysql')
    kecuali Pengecualian sebagai kesalahan:
      print("etl for {} has error".format(config['database']))
      print('pesan kesalahan: {}'.format(error))
      melanjutkan
	
  # Basis Data > sql-server
  untuk konfigurasi di sqlserver_db_config: 
    mencoba:
      print("memuat db: " + config['database'])
      etl_process(sqlserver_queries, target_cnx, config, 'sqlserver')
    kecuali Pengecualian sebagai kesalahan:
      print("etl for {} has error".format(config['database']))
      print('pesan kesalahan: {}'.format(error))
      melanjutkan

  target_cnx.close()
jika __name__ == "__main__":
  utama()

Kesimpulan

Kerja bagus! Anda telah berhasil memperoleh pemahaman dasar tentang membangun Python ETL Pipeline. Sekarang Anda dapat mengimplementasikan skrip Python ETL kustom Anda berdasarkan kebutuhan Anda dengan membuat perubahan pada database yang digunakan dan kueri yang sesuai.

Untuk menjelajahi Alat Python ETL yang banyak digunakan di industri, bacalah blog Alat Python ETL Terbaik.

Sebagian besar organisasi saat ini bekerja dengan Big Data. Oleh karena itu, membuat saluran ETL dari awal untuk data tersebut dapat memakan waktu dan menantang.

Selain itu, perusahaan perlu menginvestasikan sejumlah besar sumber daya untuk membangunnya dan kemudian menjamin bahwa mereka dapat mengikuti volume data yang tinggi dan fluktuasi skema.

Jadi, alih-alih membuat skrip ETL dari awal, Anda dapat memanfaatkan Pipa Data otomatis seperti Hevo.

Punya pemikiran tentang ini? Beri tahu kami di bawah di komentar atau bawa diskusi ke Twitter atau Facebook kami.