airflow.providers.apache.spark.operators.spark_jdbc

模組內容

類別

SparkJDBCOperator

擴展 SparkSubmitOperator 以使用 Apache Spark 執行與 JDBC 資料庫之間的資料傳輸。

class airflow.providers.apache.spark.operators.spark_jdbc.SparkJDBCOperator(*, spark_app_name='airflow-spark-jdbc', spark_conn_id='spark-default', spark_conf=None, spark_py_files=None, spark_files=None, spark_jars=None, cmd_type='spark_to_jdbc', jdbc_table=None, jdbc_conn_id='jdbc-default', jdbc_driver=None, metastore_table=None, jdbc_truncate=False, save_mode=None, save_format=None, batch_size=None, fetch_size=None, num_partitions=None, partition_column=None, lower_bound=None, upper_bound=None, create_table_column_types=None, **kwargs)[原始碼]

基於: airflow.providers.apache.spark.operators.spark_submit.SparkSubmitOperator

擴展 SparkSubmitOperator 以使用 Apache Spark 執行與 JDBC 資料庫之間的資料傳輸。

與 SparkSubmitOperator 相同,它假設 “spark-submit” 二進制檔案在 PATH 中可用。

另請參閱

有關如何使用此運算子的更多資訊,請參閱指南: SparkJDBCOperator

參數
  • spark_app_name (str) – 工作名稱 (預設 airflow-spark-jdbc)

  • spark_conn_id (str) – 在 Airflow 管理中設定的 spark 連線 ID

  • spark_conf (dict[str, Any] | None) – 任何額外的 Spark 設定屬性

  • spark_py_files (str | None) – 使用的其他 Python 檔案 (.zip, .egg 或 .py)

  • spark_files (str | None) – 要上傳到執行工作的容器的其他檔案

  • spark_jars (str | None) – 要上傳並添加到驅動程式和執行器類別路徑的其他 JAR 檔案

  • cmd_type (str) – 資料應如何流動。2 個可能的值:spark_to_jdbc:spark 從中繼資料儲存庫寫入到 jdbc 的資料 jdbc_to_spark:spark 從 jdbc 寫入到中繼資料儲存庫的資料

  • jdbc_table (str | None) – JDBC 表格的名稱

  • jdbc_conn_id (str) – 用於連接到 JDBC 資料庫的連線 ID

  • jdbc_driver (str | None) – 用於 JDBC 連線的 JDBC 驅動程式名稱。此驅動程式(通常是 jar 檔案)應在 ‘jars’ 參數中傳遞

  • metastore_table (str | None) – 中繼資料儲存庫表格的名稱,

  • jdbc_truncate (bool) – (僅限 spark_to_jdbc)Spark 是否應截斷或刪除並重新建立 JDBC 表格。這僅在 ‘save_mode’ 設定為 Overwrite 時生效。此外,如果結構描述不同,Spark 無法截斷,將刪除並重新建立

  • save_mode (str | None) – 要使用的 Spark 儲存模式(例如 overwrite、append 等)

  • save_format (str | None) – (僅限 jdbc_to_spark)要使用的 Spark 儲存格式(例如 parquet)

  • batch_size (int | None) – (僅限 spark_to_jdbc)每次往返 JDBC 資料庫要插入的批次大小。預設值為 1000

  • fetch_size (int | None) – (僅限 jdbc_to_spark)每次往返從 JDBC 資料庫要提取的批次大小。預設值取決於 JDBC 驅動程式

  • num_partitions (int | None) – Spark 可以同時使用的最大分割區數量,適用於 spark_to_jdbc 和 jdbc_to_spark 操作。這也將限制可以開啟的 JDBC 連線數量

  • partition_column (str | None) – (僅限 jdbc_to_spark)用於分割中繼資料儲存庫表格的數字欄位。如果指定,您還必須指定:num_partitions、lower_bound、upper_bound

  • lower_bound (str | None) – (僅限 jdbc_to_spark)要提取的數字分割區欄位範圍的下限。如果指定,您還必須指定:num_partitions、partition_column、upper_bound

  • upper_bound (str | None) – (僅限 jdbc_to_spark)要提取的數字分割區欄位範圍的上限。如果指定,您還必須指定:num_partitions、partition_column、lower_bound

  • create_table_column_types (str | None) – (僅限 spark_to_jdbc)建立表格時,要使用的資料庫欄位資料類型,而不是預設值。資料類型資訊應以與 CREATE TABLE 欄位語法相同的格式指定(例如:“name CHAR(64), comments VARCHAR(1024)”)。指定的類型應為有效的 Spark SQL 資料類型。

  • kwargs (Any) – 傳遞給 SparkSubmitOperator 的 kwargs。

execute(context)[原始碼]

呼叫 SparkSubmitHook 以執行提供的 Spark 工作。

on_kill()[原始碼]

覆寫此方法以在任務實例被終止時清理子進程。

在運算子內使用 threading、subprocess 或 multiprocessing 模組的任何情況都需要清理,否則會留下幽靈進程。

此條目是否有幫助?