airflow.providers.google.cloud.hooks.datafusion

此模組包含 Google DataFusion Hook。

屬性

Operation

FAILURE_STATES

SUCCESS_STATES

異常

ConflictException

捕獲 409 錯誤的異常。

類

PipelineStates

Data Fusion 流水線狀態。

DataFusionHook

Google DataFusion 的 Hook。

DataFusionAsyncHook

獲取 DataFusion 非同步 Hook 的類。

模組內容

airflow.providers.google.cloud.hooks.datafusion.Operation[原始碼]
exception airflow.providers.google.cloud.hooks.datafusion.ConflictException[原始碼]

基類: airflow.exceptions.AirflowException

捕獲 409 錯誤的異常。

class airflow.providers.google.cloud.hooks.datafusion.PipelineStates[原始碼]

Data Fusion 流水線狀態。

PENDING = 'PENDING'[原始碼]
STARTING = 'STARTING'[原始碼]
RUNNING = 'RUNNING'[原始碼]
SUSPENDED = 'SUSPENDED'[原始碼]
RESUMING = 'RESUMING'[原始碼]
COMPLETED = 'COMPLETED'[原始碼]
FAILED = 'FAILED'[原始碼]
KILLED = 'KILLED'[原始碼]
REJECTED = 'REJECTED'[原始碼]
airflow.providers.google.cloud.hooks.datafusion.FAILURE_STATES[原始碼]
airflow.providers.google.cloud.hooks.datafusion.SUCCESS_STATES[原始碼]
class airflow.providers.google.cloud.hooks.datafusion.DataFusionHook(api_version='v1beta1', gcp_conn_id='google_cloud_default', impersonation_chain=None, **kwargs)[原始碼]

基類: airflow.providers.google.common.hooks.base_google.GoogleBaseHook

Google DataFusion 的 Hook。

api_version = 'v1beta1'[原始碼]
wait_for_operation(operation)[原始碼]

等待長時間執行的操作完成。

wait_for_pipeline_state(pipeline_name, pipeline_id, instance_url, pipeline_type=DataFusionPipelineType.BATCH, namespace='default', success_states=None, failure_states=None, timeout=5 * 60)[原始碼]

輪詢流水線狀態，如果狀態失敗或超時則丟擲異常。

get_conn()[原始碼]

獲取 DataFusion 連線。

restart_instance(instance_name, location, project_id)[原始碼]

重啟單個 Data Fusion 例項。

操作完成後，例項將完全重啟。

引數:

  • instance_name (str) – 要重啟的例項名稱。

  • location (str) – 處理請求的 Cloud Data Fusion 位置。

  • project_id (str) – 例項所屬的 Google Cloud 專案 ID。

delete_instance(instance_name, location, project_id)[原始碼]

刪除單個 Data Fusion 例項。

引數:

  • instance_name (str) – 要刪除的例項名稱。

  • location (str) – 處理請求的 Cloud Data Fusion 位置。

  • project_id (str) – 例項所屬的 Google Cloud 專案 ID。

create_instance(instance_name, instance, location, project_id=PROVIDE_PROJECT_ID)[原始碼]

在指定的專案和位置建立新的 Data Fusion 例項。

引數:
get_instance(instance_name, location, project_id)[原始碼]

獲取單個 Data Fusion 例項的詳情。

引數:

  • instance_name (str) – 例項名稱。

  • location (str) – 處理請求的 Cloud Data Fusion 位置。

  • project_id (str) – 例項所屬的 Google Cloud 專案 ID。

get_instance_artifacts(instance_url, namespace='default', scope='SYSTEM')[原始碼]
patch_instance(instance_name, instance, update_mask, location, project_id=PROVIDE_PROJECT_ID)[原始碼]

更新單個 Data Fusion 例項。

引數:
create_pipeline(pipeline_name, pipeline, instance_url, namespace='default')[原始碼]

建立批次 Cloud Data Fusion 流水線。

引數:
delete_pipeline(pipeline_name, instance_url, version_id=None, namespace='default')[原始碼]

刪除批次 Cloud Data Fusion 流水線。

引數:

  • pipeline_name (str) – 您的流水線名稱。

  • version_id (str | None) – 要刪除的流水線版本

  • instance_url (str) – 例項上可訪問 REST API 的端點。

  • namespace (str) – 如果您的流水線屬於基礎版例項，則名稱空間 ID 始終為 default。如果您的流水線屬於企業版例項，則可以建立名稱空間。

list_pipelines(instance_url, artifact_name=None, artifact_version=None, namespace='default')[原始碼]

列出 Cloud Data Fusion 流水線。

引數:

  • artifact_version (str | None) – 用於過濾例項的工件版本

  • artifact_name (str | None) – 用於過濾例項的工件名稱

  • instance_url (str) – 例項上可訪問 REST API 的端點。

  • namespace (str) – 如果您的流水線屬於基礎版例項，則名稱空間 ID 始終為 default。如果您的流水線屬於企業版例項，則可以建立名稱空間。

get_pipeline_workflow(pipeline_name, instance_url, pipeline_id, pipeline_type=DataFusionPipelineType.BATCH, namespace='default')[原始碼]
start_pipeline(pipeline_name, instance_url, pipeline_type=DataFusionPipelineType.BATCH, namespace='default', runtime_args=None)[原始碼]

啟動 Cloud Data Fusion 流水線。適用於批次和流式流水線。

引數:

  • pipeline_name (str) – 您的流水線名稱。

  • pipeline_type (airflow.providers.google.cloud.utils.datafusion.DataFusionPipelineType) – 可選的流水線型別（預設為 BATCH）。

  • instance_url (str) – 例項上可訪問 REST API 的端點。

  • runtime_args (dict[str, Any] | None) – 要傳遞給流水線的可選執行時 JSON 引數

  • namespace (str) – 如果您的流水線屬於基礎版例項，則名稱空間 ID 始終為 default。如果您的流水線屬於企業版例項，則可以建立名稱空間。

stop_pipeline(pipeline_name, instance_url, namespace='default')[原始碼]

停止 Cloud Data Fusion 流水線。適用於批次和流式流水線。

引數:

  • pipeline_name (str) – 您的流水線名稱。

  • instance_url (str) – 例項上可訪問 REST API 的端點。

  • namespace (str) – 如果您的流水線屬於基礎版例項，則名稱空間 ID 始終為 default。如果您的流水線屬於企業版例項，則可以建立名稱空間。

static cdap_program_type(pipeline_type)[原始碼]

根據流水線型別檢索 CDAP 程式型別。

引數:

pipeline_type (airflow.providers.google.cloud.utils.datafusion.DataFusionPipelineType) – 流水線型別。

static cdap_program_id(pipeline_type)[原始碼]

根據流水線型別檢索 CDAP 程式 ID。

引數:

pipeline_type (airflow.providers.google.cloud.utils.datafusion.DataFusionPipelineType) – 流水線型別。

class airflow.providers.google.cloud.hooks.datafusion.DataFusionAsyncHook(**kwargs)[原始碼]

基類: airflow.providers.google.common.hooks.base_google.GoogleBaseAsyncHook

獲取 DataFusion 非同步 Hook 的類。

sync_hook_class[原始碼]
scopes = ['https://www.googleapis.com/auth/cloud-platform'][原始碼]
async get_pipeline(instance_url, namespace, pipeline_name, pipeline_id, session, pipeline_type=DataFusionPipelineType.BATCH)[原始碼]
async get_pipeline_status(pipeline_name, instance_url, pipeline_id, pipeline_type=DataFusionPipelineType.BATCH, namespace='default', success_states=None)[原始碼]

非同步獲取 Cloud Data Fusion 流水線狀態。

引數:

  • pipeline_name (str) – 您的流水線名稱。

  • instance_url (str) – 例項上可訪問 REST API 的端點。

  • pipeline_id (str) – 與特定流水線關聯的唯一流水線 ID。

  • pipeline_type (airflow.providers.google.cloud.utils.datafusion.DataFusionPipelineType) – 可選的流水線型別（預設為 batch）。

  • namespace (str) – 如果您的流水線屬於基礎版例項，則名稱空間 ID 始終為 default。如果您的流水線屬於企業版例項，則可以建立名稱空間。

  • success_states (list[str] | None) – 如果提供，operator 將等待流水線進入提供的狀態之一。

本條目是否有幫助？