Làm cách nào để lấy và xử lý các bản ghi mysql bằng Airflow?

Tôi đã thực sự đấu tranh với điều này trong 90 phút qua, đây là một cách dễ hiểu hơn để làm theo cho những người mới:

from airflow.hooks.mysql_hook import MySqlHook

def fetch_records():
  request = "SELECT * FROM your_table"
  mysql_hook = MySqlHook(mysql_conn_id = 'the_connection_name_sourced_from_the_ui', schema = 'specific_db')
  connection = mysql_hook.get_conn()
  cursor = connection.cursor()
  cursor.execute(request)
  sources = cursor.fetchall()
  print(sources)

...your DAG() as dag: code

task = PythonOperator(
  task_id = 'fetch_records',
  python_callable = fetch_records
)

Điều này trả về nhật ký nội dung của truy vấn DB của bạn.

Tôi hy vọng điều này có thể sử dụng cho người khác.