28 集成 · Airflow中文文檔

# 集成 * [反向代理](28) * [Azure：Microsoft Azure](28) * [AWS：亞馬遜網絡服務](28) * [Databricks](28) * [GCP：Google云端平臺](28) ## 反向代理可以在反向代理后面設置氣流，并能夠靈活地設置其端點。例如，您可以配置反向代理以獲取： ``` https : // lab . mycompany . com / myorg / airflow / ``` 為此，您需要在<cite>airflow.cfg中</cite>設置以下設置： ``` base_url = http : // my_host / myorg / airflow ``` 此外，如果您使用Celery Executor，您可以使用以下命令獲取<cite>/</cite> in <cite>myorg / flower中的</cite> Flower： ``` flower_url_prefix = / myorg / flower ``` 您的反向代理（例如：nginx）應配置如下： * 傳遞url和http標頭作為Airflow網絡服務器，沒有任何重寫，例如： ``` server { listen 80; server_name lab.mycompany.com; location /myorg/airflow/ { proxy_pass http://localhost:8080; proxy_set_header Host $host; proxy_redirect off; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; } } ``` * 重寫花端點的url： ``` server { listen 80; server_name lab.mycompany.com; location /myorg/flower/ { rewrite ^/myorg/flower/(.*)$ /$1 break; # remove prefix from http header proxy_pass http://localhost:5555; proxy_set_header Host $host; proxy_redirect off; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; } } ``` ## Azure：Microsoft Azure Airflow對Microsoft Azure的支持有限：僅存在Azure Blob存儲和Azure Data Lake的接口。 Blob存儲的鉤子，傳感器和操作員以及Azure Data Lake Hook都在contrib部分。 ### Azure Blob存儲所有類都通過Window Azure Storage Blob協議進行通信。確保存在類型為<cite>wasb</cite>的Airflow連接。可以通過在額外字段中提供登錄（=存儲帳戶名稱）和密碼（= KEY），或登錄和SAS令牌來完成授權（有關<cite>示例</cite> ，請參閱連接<cite>wasb_default</cite> ）。 * [WasbBlobSensor](28) ：檢查Azure Blob存儲上是否存在blob。 * [WasbPrefixSensor](28) ：檢查Azure Blob存儲上是否存在與前綴匹配的blob。 * [FileToWasbOperator](28) ：將本地文件作為blob [上載到](28)容器。 * [WasbHook](28) ：與Azure Blob存儲的接口。 #### WasbBlobSensor ``` class airflow.contrib.sensors.wasb_sensor.WasbBlobSensor(container_name, blob_name, wasb_conn_id='wasb_default', check_options=None, *args, **kwargs) ``` 基礎： [`airflow.sensors.base_sensor_operator.BaseSensorOperator`](code.html "airflow.sensors.base_sensor_operator.BaseSensorOperator") 等待blob到達Azure Blob存儲。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **container_name** （ _str_ ） - 容器的名稱。 * **blob_name** （ _str_ ） - blob的名稱。 * **wasb_conn_id** （ _str_ ） - 對wasb連接的引用。 * **check_options** （ _dict_ ） - <cite>WasbHook.check_for_blob（）</cite>采用的可選關鍵字參數。 | | --- | --- | ``` poke(context) ``` 傳感器在派生此類時定義的功能應該覆蓋。 #### WasbPrefixSensor ``` class airflow.contrib.sensors.wasb_sensor.WasbPrefixSensor(container_name, prefix, wasb_conn_id='wasb_default', check_options=None, *args, **kwargs) ``` 基礎： [`airflow.sensors.base_sensor_operator.BaseSensorOperator`](code.html "airflow.sensors.base_sensor_operator.BaseSensorOperator") 等待與前綴匹配的blob到達Azure Blob存儲。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **container_name** （ _str_ ） - 容器的名稱。 * **prefix** （ _str_ ） - blob的前綴。 * **wasb_conn_id** （ _str_ ） - 對wasb連接的引用。 * **check_options** （ _dict_ ） - <cite>WasbHook.check_for_prefix（）</cite>采用的可選關鍵字參數。 | | --- | --- | ``` poke(context) ``` 傳感器在派生此類時定義的功能應該覆蓋。 #### FileToWasbOperator ``` class airflow.contrib.operators.file_to_wasb.FileToWasbOperator(file_path, container_name, blob_name, wasb_conn_id='wasb_default', load_options=None, *args, **kwargs) ``` 基礎： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 將文件上載到Azure Blob存儲。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **file_path** （ _str_ ） - 要加載的文件的路徑。（模板） * **container_name** （ _str_ ） - 容器的名稱。（模板） * **blob_name** （ _str_ ） - blob的名稱。（模板） * **wasb_conn_id** （ _str_ ） - 對wasb連接的引用。 * **load_options** （ _dict_ ） - <cite>WasbHook.load_file（）</cite>采用的可選關鍵字參數。 | | --- | --- | ``` execute(context) ``` 將文件上載到Azure Blob存儲。 #### WasbHook ``` class airflow.contrib.hooks.wasb_hook.WasbHook(wasb_conn_id='wasb_default') ``` 基礎： `airflow.hooks.base_hook.BaseHook` 通過wasb：//協議與Azure Blob存儲進行交互。在連接的“額外”字段中傳遞的其他選項將傳遞給<cite>BlockBlockService（）</cite>構造函數。例如，通過添加{“sas_token”：“YOUR_TOKEN”}使用SAS令牌進行身份驗證。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **wasb_conn_id** （ _str_ ） - 對wasb連接的引用。 | | --- | --- | ``` check_for_blob(container_name, blob_name, **kwargs) ``` 檢查Azure Blob存儲上是否存在Blob。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **container_name** （ _str_ ） - 容器的名稱。 * **blob_name** （ _str_ ） - blob的名稱。 * **kwargs** （ _object_ ） - <cite>BlockBlobService.exists（）</cite>采用的可選關鍵字參數。 | | --- | --- | | 返回： | 如果blob存在則為True，否則為False。 | | --- | --- | ：rtype布爾 ``` check_for_prefix(container_name, prefix, **kwargs) ``` 檢查Azure Blob存儲上是否存在前綴。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **container_name** （ _str_ ） - 容器的名稱。 * **prefix** （ _str_ ） - blob的前綴。 * **kwargs** （ _object_ ） - <cite>BlockBlobService.list_blobs（）</cite>采用的可選關鍵字參數。 | | --- | --- | | 返回： | 如果存在與前綴匹配的blob，則為True，否則為False。 | | --- | --- | ：rtype布爾 ``` get_conn() ``` 返回BlockBlobService對象。 ``` get_file(file_path, container_name, blob_name, **kwargs) ``` 從Azure Blob存儲下載文件。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **file_path** （ _str_ ） - 要下載的文件的路徑。 * **container_name** （ _str_ ） - 容器的名稱。 * **blob_name** （ _str_ ） - blob的名稱。 * **kwargs** （ _object_ ） - <cite>BlockBlobService.create_blob_from_path（）</cite>采用的可選關鍵字參數。 | | --- | --- | ``` load_file(file_path, container_name, blob_name, **kwargs) ``` 將文件上載到Azure Blob存儲。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **file_path** （ _str_ ） - 要加載的文件的路徑。 * **container_name** （ _str_ ） - 容器的名稱。 * **blob_name** （ _str_ ） - blob的名稱。 * **kwargs** （ _object_ ） - <cite>BlockBlobService.create_blob_from_path（）</cite>采用的可選關鍵字參數。 | | --- | --- | ``` load_string(string_data, container_name, blob_name, **kwargs) ``` 將字符串上載到Azure Blob存儲。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **string_data** （ _str_ ） - 要加載的字符串。 * **container_name** （ _str_ ） - 容器的名稱。 * **blob_name** （ _str_ ） - blob的名稱。 * **kwargs** （ _object_ ） - <cite>BlockBlobService.create_blob_from_text（）</cite>采用的可選關鍵字參數。 | | --- | --- | ``` read_file(container_name, blob_name, **kwargs) ``` 從Azure Blob Storage讀取文件并以字符串形式返回。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **container_name** （ _str_ ） - 容器的名稱。 * **blob_name** （ _str_ ） - blob的名稱。 * **kwargs** （ _object_ ） - <cite>BlockBlobService.create_blob_from_path（）</cite>采用的可選關鍵字參數。 | | --- | --- | ### Azure文件共享 SMB文件共享的云變體。確保存在類型為<cite>wasb</cite>的Airflow連接。可以通過在額外字段中提供登錄（=存儲帳戶名稱）和密碼（=存儲帳戶密鑰）或登錄和SAS令牌來完成授權（有關<cite>示例</cite> ，請參閱連接<cite>wasb_default</cite> ）。 #### AzureFileShareHook ``` class airflow.contrib.hooks.azure_fileshare_hook.AzureFileShareHook(wasb_conn_id='wasb_default') ``` 基礎： `airflow.hooks.base_hook.BaseHook` 與Azure FileShare存儲交互。在連接的“額外”字段中傳遞的其他選項將傳遞給<cite>FileService（）</cite>構造函數。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **wasb_conn_id** （ _str_ ） - 對wasb連接的引用。 | | --- | --- | ``` check_for_directory(share_name, directory_name, **kwargs) ``` 檢查Azure文件共享上是否存在目錄。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **share_name** （ _str_ ） - 共享的名稱。 * **directory_name** （ _str_ ） - 目錄的名稱。 * **kwargs** （ _object_ ） - <cite>FileService.exists（）</cite>采用的可選關鍵字參數。 | | --- | --- | | 返回： | 如果文件存在則為True，否則為False。 | | --- | --- | ：rtype布爾 ``` check_for_file(share_name, directory_name, file_name, **kwargs) ``` 檢查Azure文件共享上是否存在文件。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **share_name** （ _str_ ） - 共享的名稱。 * **directory_name** （ _str_ ） - 目錄的名稱。 * **file_name** （ _str_ ） - 文件名。 * **kwargs** （ _object_ ） - <cite>FileService.exists（）</cite>采用的可選關鍵字參數。 | | --- | --- | | 返回： | 如果文件存在則為True，否則為False。 | | --- | --- | ：rtype布爾 ``` create_directory(share_name, directory_name, **kwargs) ``` 在Azure文件共享上創建新的目標。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **share_name** （ _str_ ） - 共享的名稱。 * **directory_name** （ _str_ ） - 目錄的名稱。 * **kwargs** （ _object_ ） - <cite>FileService.create_directory（）</cite>采用的可選關鍵字參數。 | | --- | --- | | 返回： | 文件和目錄列表 | | --- | --- | ：rtype列表 ``` get_conn() ``` 返回FileService對象。 ``` get_file(file_path, share_name, directory_name, file_name, **kwargs) ``` 從Azure文件共享下載文件。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **file_path** （ _str_ ） - 存儲文件的位置。 * **share_name** （ _str_ ） - 共享的名稱。 * **directory_name** （ _str_ ） - 目錄的名稱。 * **file_name** （ _str_ ） - 文件名。 * **kwargs** （ _object_ ） - <cite>FileService.get_file_to_path（）</cite>采用的可選關鍵字參數。 | | --- | --- | ``` get_file_to_stream(stream, share_name, directory_name, file_name, **kwargs) ``` 從Azure文件共享下載文件。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **stream** （類_文件對象_ ） - 用于存儲文件的文件句柄。 * **share_name** （ _str_ ） - 共享的名稱。 * **directory_name** （ _str_ ） - 目錄的名稱。 * **file_name** （ _str_ ） - 文件名。 * **kwargs** （ _object_ ） - <cite>FileService.get_file_to_stream（）</cite>采用的可選關鍵字參數。 | | --- | --- | ``` list_directories_and_files(share_name, directory_name=None, **kwargs) ``` 返回存儲在Azure文件共享中的目錄和文件列表。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **share_name** （ _str_ ） - 共享的名稱。 * **directory_name** （ _str_ ） - 目錄的名稱。 * **kwargs** （ _object_ ） - <cite>FileService.list_directories_and_files（）</cite>采用的可選關鍵字參數。 | | --- | --- | | 返回： | 文件和目錄列表 | | --- | --- | ：rtype列表 ``` load_file(file_path, share_name, directory_name, file_name, **kwargs) ``` 將文件上載到Azure文件共享。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **file_path** （ _str_ ） - 要加載的文件的路徑。 * **share_name** （ _str_ ） - 共享的名稱。 * **directory_name** （ _str_ ） - 目錄的名稱。 * **file_name** （ _str_ ） - 文件名。 * **kwargs** （ _object_ ） - <cite>FileService.create_file_from_path（）</cite>采用的可選關鍵字參數。 | | --- | --- | ``` load_stream(stream, share_name, directory_name, file_name, count, **kwargs) ``` 將流上載到Azure文件共享。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **stream** （類_文件_ ） - 打開的文件/流作為文件內容上傳。 * **share_name** （ _str_ ） - 共享的名稱。 * **directory_name** （ _str_ ） - 目錄的名稱。 * **file_name** （ _str_ ） - 文件名。 * **count** （ _int_ ） - 流的大小（以字節為單位） * **kwargs** （ _object_ ） - <cite>FileService.create_file_from_stream（）</cite>采用的可選關鍵字參數。 | | --- | --- | ``` load_string(string_data, share_name, directory_name, file_name, **kwargs) ``` 將字符串上載到Azure文件共享。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **string_data** （ _str_ ） - 要加載的字符串。 * **share_name** （ _str_ ） - 共享的名稱。 * **directory_name** （ _str_ ） - 目錄的名稱。 * **file_name** （ _str_ ） - 文件名。 * **kwargs** （ _object_ ） - <cite>FileService.create_file_from_text（）</cite>采用的可選關鍵字參數。 | | --- | --- | ### 記錄可以將Airflow配置為在Azure Blob存儲中讀取和寫入任務日志。請參閱[將日志寫入Azure Blob存儲](howto/write-logs.html) 。 ### Azure Data Lake AzureDataLakeHook通過與WebHDFS兼容的REST API進行通信。確保存在<cite>azure_data_lake</cite>類型的氣流連接。可以通過提供登錄（=客戶端ID），密碼（=客戶端密鑰）和額外字段租戶（租戶）和account_name（帳戶名稱）來完成授權 > （有關<cite>示例</cite> ，請參閱<cite>azure_data_lake_default</cite>連接）。 * [AzureDataLakeHook](28) ：與Azure Data Lake的接口。 #### AzureDataLakeHook ``` class airflow.contrib.hooks.azure_data_lake_hook.AzureDataLakeHook(azure_data_lake_conn_id='azure_data_lake_default') ``` 基礎： `airflow.hooks.base_hook.BaseHook` 與Azure Data Lake進行交互。客戶端ID和客戶端密鑰應該在用戶和密碼參數中。租戶和帳戶名稱應為{“租戶”：“<TENANT>”，“account_name”：“ACCOUNT_NAME”}的額外字段。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **azure_data_lake_conn_id** （ _str_ ） - 對Azure Data Lake連接的引用。 | | --- | --- | ``` check_for_file(file_path) ``` 檢查Azure Data Lake上是否存在文件。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **file_path** （ _str_ ） - 文件的路徑和名稱。 | | --- | --- | | 返回： | 如果文件存在則為True，否則為False。 | | --- | --- | ：rtype布爾 ``` download_file(local_path, remote_path, nthreads=64, overwrite=True, buffersize=4194304, blocksize=4194304) ``` 從Azure Blob存儲下載文件。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **local_path** （ _str_ ） - 本地路徑。如果下載單個文件，將寫入此特定文件，除非它是現有目錄，在這種情況下，將在其中創建文件。如果下載多個文件，這是要寫入的根目錄。將根據需要創建目錄。 * **remote_path** （ _str_ ） - 用于查找遠程文件的遠程路徑/ globstring。不支持使用<cite>**的</cite>遞歸glob模式。 * **nthreads** （ _int_ ） - 要使用的線程數。如果為None，則使用核心數。 * **overwrite** （ _bool_ ） - 是否強制覆蓋現有文件/目錄。如果False和遠程路徑是目錄，則無論是否覆蓋任何文件都將退出。如果為True，則實際僅覆蓋匹配的文件名。 * **buffersize** （ _int_ ） - int [2 ** 22]內部緩沖區的字節數。此塊不能大于塊，并且不能小于塊。 * **blocksize** （ _int_ ） - int [2 ** 22]塊的字節數。在每個塊中，我們為每個API調用編寫一個較小的塊。這個塊不能大于塊。 | | --- | --- | ``` get_conn() ``` 返回AzureDLFileSystem對象。 ``` upload_file(local_path, remote_path, nthreads=64, overwrite=True, buffersize=4194304, blocksize=4194304) ``` 將文件上載到Azure Data Lake。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **local_path** （ _str_ ） - 本地路徑。可以是單個文件，目錄（在這種情況下，遞歸上傳）或glob模式。不支持使用<cite>**的</cite>遞歸glob模式。 * **remote_path** （ _str_ ） - 要上傳的遠程路徑; 如果有多個文件，這就是要寫入的dircetory根目錄。 * **nthreads** （ _int_ ） - 要使用的線程數。如果為None，則使用核心數。 * **overwrite** （ _bool_ ） - 是否強制覆蓋現有文件/目錄。如果False和遠程路徑是目錄，則無論是否覆蓋任何文件都將退出。如果為True，則實際僅覆蓋匹配的文件名。 * **buffersize** （ _int_ ） - int [2 ** 22]內部緩沖區的字節數。此塊不能大于塊，并且不能小于塊。 * **blocksize** （ _int_ ） - int [2 ** 22]塊的字節數。在每個塊中，我們為每個API調用編寫一個較小的塊。這個塊不能大于塊。 | | --- | --- | ## AWS：亞馬遜網絡服務 Airflow廣泛支持Amazon Web Services。但請注意，Hook，Sensors和Operators都在contrib部分。 ### AWS EMR * [EmrAddStepsOperator](28) ：向現有EMR JobFlow添加步驟。 * [EmrCreateJobFlowOperator](28) ：創建EMR JobFlow，從EMR連接讀取配置。 * [EmrTerminateJobFlowOperator](28) ：終止EMR JobFlow。 * [EmrHook](28) ：與AWS EMR互動。 #### EmrAddStepsOperator ``` class airflow.contrib.operators.emr_add_steps_operator.EmrAddStepsOperator(job_flow_id, aws_conn_id='s3_default', steps=None, *args, **kwargs) ``` 基礎： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 向現有EMR job_flow添加步驟的運算符。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **job_flow_id** - 要添加步驟的JobFlow的ID。（模板） * **aws_conn_id** （ _str_ ） - 與使用的aws連接 * **步驟** （ _列表_ ） - 要添加到作業流的boto3樣式步驟。（模板） | | --- | --- | #### EmrCreateJobFlowOperator ``` class airflow.contrib.operators.emr_create_job_flow_operator.EmrCreateJobFlowOperator(aws_conn_id='s3_default', emr_conn_id='emr_default', job_flow_overrides=None, *args, **kwargs) ``` 基礎： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 創建EMR JobFlow，從EMR連接讀取配置。可以傳遞JobFlow覆蓋的字典，覆蓋連接中的配置。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **aws_conn_id** （ _str_ ） - 與使用的aws連接 * **emr_conn_id** （ _str_ ） - 要使用的emr連接 * **job_flow_overrides** - 用于覆蓋emr_connection extra的boto3樣式參數。（模板） | | --- | --- | #### EmrTerminateJobFlowOperator ``` class airflow.contrib.operators.emr_terminate_job_flow_operator.EmrTerminateJobFlowOperator(job_flow_id, aws_conn_id='s3_default', *args, **kwargs) ``` 基礎： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 運營商終止EMR JobFlows。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **job_flow_id** - 要終止的JobFlow的id。（模板） * **aws_conn_id** （ _str_ ） - 與使用的aws連接 | | --- | --- | #### EmrHook ``` class airflow.contrib.hooks.emr_hook.EmrHook(emr_conn_id=None, *args, **kwargs) ``` 基地： [`airflow.contrib.hooks.aws_hook.AwsHook`](code.html "airflow.contrib.hooks.aws_hook.AwsHook") 與AWS EMR交互。 emr_conn_id只是使用create_job_flow方法所必需的。 ``` create_job_flow(job_flow_overrides) ``` 使用EMR連接中的配置創建作業流。 json額外哈希的鍵可以具有boto3 run_job_flow方法的參數。此配置的覆蓋可以作為job_flow_overrides傳遞。 ### AWS S3 * [S3Hook](28) ：與AWS S3交互。 * [S3FileTransformOperator](28) ：將數據從源S3位置復制到本地文件系統上的臨時位置。 * [S3ListOperator](28) ：列出與S3位置的鍵前綴匹配的文件。 * [S3ToGoogleCloudStorageOperator](28) ：將S3位置與Google云端存儲[分區](28)同步。 * [S3ToHiveTransfer](28) ：將數據從S3移動到Hive。操作員從S3下載文件，在將文件加載到Hive表之前將其存儲在本地。 #### S3Hook ``` class airflow.hooks.S3_hook.S3Hook(aws_conn_id='aws_default') ``` 基地： [`airflow.contrib.hooks.aws_hook.AwsHook`](code.html "airflow.contrib.hooks.aws_hook.AwsHook") 使用boto3庫與AWS S3交互。 ``` check_for_bucket(bucket_name) ``` 檢查bucket_name是否存在。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **bucket_name** （ _str_ ） - 存儲桶的名稱 | | --- | --- | ``` check_for_key(key, bucket_name=None) ``` 檢查存儲桶中是否存在密鑰 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **key** （ _str_ ） - 指向文件的S3鍵 * **bucket_name** （ _str_ ） - 存儲文件的存儲桶的名稱 | | --- | --- | ``` check_for_prefix(bucket_name, prefix, delimiter) ``` 檢查存儲桶中是否存在前綴 ``` check_for_wildcard_key(wildcard_key, bucket_name=None, delimiter='') ``` 檢查桶中是否存在與通配符表達式匹配的密鑰 ``` get_bucket(bucket_name) ``` 返回boto3.S3.Bucket對象 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **bucket_name** （ _str_ ） - 存儲桶的名稱 | | --- | --- | ``` get_key(key, bucket_name=None) ``` 返回boto3.s3.Object <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **key** （ _str_ ） - 密鑰的路徑 * **bucket_name** （ _str_ ） - 存儲桶的名稱 | | --- | --- | ``` get_wildcard_key(wildcard_key, bucket_name=None, delimiter='') ``` 返回與通配符表達式匹配的boto3.s3.Object對象 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **wildcard_key** （ _str_ ） - 密鑰的路徑 * **bucket_name** （ _str_ ） - 存儲桶的名稱 | | --- | --- | ``` list_keys(bucket_name, prefix='', delimiter='', page_size=None, max_items=None) ``` 列出前綴下的存儲桶中的密鑰，但不包含分隔符 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket_name** （ _str_ ） - 存儲桶的名稱 * **prefix** （ _str_ ） - 一個密鑰前綴 * **delimiter** （ _str_ ） - 分隔符標記鍵層次結構。 * **page_size** （ _int_ ） - 分頁大小 * **max_items** （ _int_ ） - 要返回的最大項目數 | | --- | --- | ``` list_prefixes(bucket_name, prefix='', delimiter='', page_size=None, max_items=None) ``` 列出前綴下的存儲桶中的前綴 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket_name** （ _str_ ） - 存儲桶的名稱 * **prefix** （ _str_ ） - 一個密鑰前綴 * **delimiter** （ _str_ ） - 分隔符標記鍵層次結構。 * **page_size** （ _int_ ） - 分頁大小 * **max_items** （ _int_ ） - 要返回的最大項目數 | | --- | --- | ``` load_bytes(bytes_data, key, bucket_name=None, replace=False, encrypt=False) ``` 將字節加載到S3 這是為了方便在S3中刪除字符串。它使用boto基礎結構將文件發送到s3。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bytes_data** （ _bytes_ ） - 設置為密鑰內容的字節。 * **key** （ _str_ ） - 指向文件的S3鍵 * **bucket_name** （ _str_ ） - 存儲文件的存儲桶的名稱 * **replace** （ _bool_ ） - 一個標志，用于決定是否覆蓋密鑰（如果已存在） * **encrypt** （ _bool_ ） - 如果為True，則文件將在服務器端由S3加密，并在S3中靜止時以加密形式存儲。 | | --- | --- | ``` load_file(filename, key, bucket_name=None, replace=False, encrypt=False) ``` 將本地文件加載到S3 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **filename** （ _str_ ） - 要加載的文件的名稱。 * **key** （ _str_ ） - 指向文件的S3鍵 * **bucket_name** （ _str_ ） - 存儲文件的存儲桶的名稱 * **replace** （ _bool_ ） - 一個標志，用于決定是否覆蓋密鑰（如果已存在）。如果replace為False且密鑰存在，則會引發錯誤。 * **encrypt** （ _bool_ ） - 如果為True，則文件將在服務器端由S3加密，并在S3中靜止時以加密形式存儲。 | | --- | --- | ``` load_string(string_data, key, bucket_name=None, replace=False, encrypt=False, encoding='utf-8') ``` 將字符串加載到S3 這是為了方便在S3中刪除字符串。它使用boto基礎結構將文件發送到s3。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **string_data** （ _str_ ） - 要設置為鍵的內容的字符串。 * **key** （ _str_ ） - 指向文件的S3鍵 * **bucket_name** （ _str_ ） - 存儲文件的存儲桶的名稱 * **replace** （ _bool_ ） - 一個標志，用于決定是否覆蓋密鑰（如果已存在） * **encrypt** （ _bool_ ） - 如果為True，則文件將在服務器端由S3加密，并在S3中靜止時以加密形式存儲。 | | --- | --- | ``` read_key(key, bucket_name=None) ``` 從S3讀取密鑰 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **key** （ _str_ ） - 指向文件的S3鍵 * **bucket_name** （ _str_ ） - 存儲文件的存儲桶的名稱 | | --- | --- | ``` select_key(key, bucket_name=None, expression='SELECT * FROM S3Object', expression_type='SQL', input_serialization={'CSV': {}}, output_serialization={'CSV': {}}) ``` 使用S3 Select讀取密鑰。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **key** （ _str_ ） - 指向文件的S3鍵 * **bucket_name** （ _str_ ） - 存儲文件的存儲桶的名稱 * **expression** （ _str_ ） - S3選擇表達式 * **expression_type** （ _str_ ） - S3選擇表達式類型 * **input_serialization** （ _dict_ ） - S3選擇輸入數據序列化格式 * **output_serialization** （ _dict_ ） - S3選擇輸出數據序列化格式 | | --- | --- | | 返回： | 通過S3 Select檢索原始數據的子集 | | --- | --- | | 返回類型： | 海峽 | | --- | --- | 也可以看看有關S3 Select參數的更多詳細信息： [http](http://boto3.readthedocs.io/en/latest/reference/services/s3.html) ： [//boto3.readthedocs.io/en/latest/reference/services/s3.html#S3.Client.select_object_content](http://boto3.readthedocs.io/en/latest/reference/services/s3.html) #### S3FileTransformOperator ``` class airflow.operators.s3_file_transform_operator.S3FileTransformOperator(source_s3_key, dest_s3_key, transform_script=None, select_expression=None, source_aws_conn_id='aws_default', dest_aws_conn_id='aws_default', replace=False, *args, **kwargs) ``` 基礎： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 將數據從源S3位置復制到本地文件系統上的臨時位置。根據轉換腳本的指定對此文件運行轉換，并將輸出上載到目標S3位置。本地文件系統中的源文件和目標文件的位置作為轉換腳本的第一個和第二個參數提供。轉換腳本應該從源讀取數據，轉換它并將輸出寫入本地目標文件。然后，操作員接管控制并將本地目標文件上載到S3。 S3 Select也可用于過濾源內容。如果指定了S3 Select表達式，則用戶可以省略轉換腳本。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **source_s3_key** （ _str_ ） - 從S3檢索的密鑰。（模板） * **source_aws_conn_id** （ _str_ ） - 源s3連接 * **dest_s3_key** （ _str_ ） - 從S3寫入的密鑰。（模板） * **dest_aws_conn_id** （ _str_ ） - 目標s3連接 * **replace** （ _bool_ ） - 替換dest S3密鑰（如果已存在） * **transform_script** （ _str_ ） - 可執行轉換腳本的位置 * **select_expression** （ _str_ ） - S3選擇表達式 | | --- | --- | #### S3ListOperator ``` class airflow.contrib.operators.s3_list_operator.S3ListOperator(bucket, prefix='', delimiter='', aws_conn_id='aws_default', *args, **kwargs) ``` 基礎： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 列出桶中具有名稱中給定字符串前綴的所有對象。此運算符返回一個python列表，其中包含可由<cite>xcom</cite>在下游任務中使用的對象名稱。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket** （ _string_ ） - S3存儲桶在哪里找到對象。（模板） * **prefix** （ _string_ ） - 用于過濾名稱以此前綴開頭的對象的前綴字符串。（模板） * **delimiter** （ _string_ ） - 分隔符標記鍵層次結構。（模板） * **aws_conn_id** （ _string_ ） - 連接到S3存儲時使用的連接ID。 | | --- | --- | ``` Example: ``` 以下運算符將列出`data`存儲區中S3 `customers/2018/04/` key的所有文件（不包括子文件夾）。 ``` s3_file = S3ListOperator ( task_id = 'list_3s_files' , bucket = 'data' , prefix = 'customers/2018/04/' , delimiter = '/' , aws_conn_id = 'aws_customers_conn' ) ``` #### S3ToGoogleCloudStorageOperator ``` class airflow.contrib.operators.s3_to_gcs_operator.S3ToGoogleCloudStorageOperator(bucket, prefix='', delimiter='', aws_conn_id='aws_default', dest_gcs_conn_id=None, dest_gcs=None, delegate_to=None, replace=False, *args, **kwargs) ``` 基礎： [`airflow.contrib.operators.s3_list_operator.S3ListOperator`](28 "airflow.contrib.operators.s3_list_operator.S3ListOperator") 將S3密鑰（可能是前綴）與Google云端存儲目標路徑同步。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket** （ _string_ ） - S3存儲桶在哪里找到對象。（模板） * **prefix** （ _string_ ） - 前綴字符串，用于過濾名稱以此前綴開頭的對象。（模板） * **delimiter** （ _string_ ） - 分隔符標記鍵層次結構。（模板） * **aws_conn_id** （ _string_ ） - 源S3連接 * **dest_gcs_conn_id** （ _字符串_ ） - 連接到Google云端存儲時要使用的目標連接ID。 * **dest_gcs** （ _字符串_ ） - 要存儲文件的目標Google云端存儲**分區**和前綴。（模板） * **delegate_to** （ _string_ ） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **replace** （ _bool_ ） - 是否要替換現有目標文件。 | | --- | --- | **示例** ：.. code-block :: python > ``` > s3_to_gcs_op = S3ToGoogleCloudStorageOperator( > ``` > > task_id ='s3_to_gcs_example'，bucket ='my-s3-bucket'，prefix ='data / customers-201804'，dest_gcs_conn_id ='google_cloud_default'，dest_gcs ='gs：//my.gcs.bucket/some/customers/' ，replace = False，dag = my-dag）請注意， `bucket` ， `prefix` ， `delimiter`和`dest_gcs`是模板化的，因此如果您愿意，可以在其中使用變量。 #### S3ToHiveTransfer ``` class airflow.operators.s3_to_hive_operator.S3ToHiveTransfer(s3_key, field_dict, hive_table, delimiter=', ', create=True, recreate=False, partition=None, headers=False, check_headers=False, wildcard_match=False, aws_conn_id='aws_default', hive_cli_conn_id='hive_cli_default', input_compressed=False, tblproperties=None, select_expression=None, *args, **kwargs) ``` 基礎： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 將數據從S3移動到Hive。操作員從S3下載文件，在將文件加載到Hive表之前將其存儲在本地。如果`create`或`recreate`參數設置為`True` ，則會生成`CREATE TABLE`和`DROP TABLE`語句。 Hive數據類型是從游標的元數據中推斷出來的。請注意，Hive中生成的表使用`STORED AS textfile` ，這不是最有效的序列化格式。如果加載了大量數據和/或表格被大量查詢，您可能只想使用此運算符將數據暫存到臨時表中，然后使用`HiveOperator`將其加載到最終目標中。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **s3_key** （ _str_ ） - 從S3檢索的密鑰。（模板） * **field_dict** （ _dict_ ） - 字段的字典在文件中命名為鍵，其Hive類型為值 * **hive_table** （ _str_ ） - 目標Hive表，使用點表示法來定位特定數據庫。（模板） * **create** （ _bool_ ） - 是否創建表，如果它不存在 * **recreate** （ _bool_ ） - 是否在每次執行時刪除并重新創建表 * **partition** （ _dict_ ） - 將目標分區作為分區列和值的字典。（模板） * **headers** （ _bool_ ） - 文件是否包含第一行的列名 * **check_headers** （ _bool_ ） - 是否應該根據field_dict的鍵檢查第一行的列名 * **wildcard_match** （ _bool_ ） - 是否應將s3_key解釋為Unix通配符模式 * **delimiter** （ _str_ ） - 文件中的字段分隔符 * **aws_conn_id** （ _str_ ） - 源s3連接 * **hive_cli_conn_id** （ _str_ ） - 目標配置單元連接 * **input_compressed** （ _bool_ ） - 布爾值，用于確定是否需要文件解壓縮來處理標頭 * **tblproperties** （ _dict_ ） - 正在創建的hive表的TBLPROPERTIES * **select_expression** （ _str_ ） - S3選擇表達式 | | --- | --- | ### AWS EC2容器服務 * [ECSOperator](28) ：在AWS EC2容器服務上執行任務。 #### ECSOperator ``` class airflow.contrib.operators.ecs_operator.ECSOperator(task_definition, cluster, overrides, aws_conn_id=None, region_name=None, launch_type='EC2', **kwargs) ``` 基礎： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 在AWS EC2 Container Service上執行任務 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **task_definition** （ _str_ ） - EC2容器服務上的任務定義名稱 * **cluster** （ _str_ ） - EC2 Container Service上的群集名稱 * **aws_conn_id** （ _str_ ） - AWS憑證/區域名稱的連接ID。如果為None，將使用憑證boto3策略（ [http://boto3.readthedocs.io/en/latest/guide/configuration.html](http://boto3.readthedocs.io/en/latest/guide/configuration.html) ）。 * **region_name** - 要在AWS Hook中使用的區域名稱。覆蓋連接中的region_name（如果提供） * **launch_type** - 運行任務的啟動類型（'EC2'或'FARGATE'） | | --- | --- | | 帕拉姆： | 覆蓋：boto3將接收的相同參數（模板化）： [http](http://boto3.readthedocs.org/en/latest/reference/services/ecs.html) ：//boto3.readthedocs.org/en/latest/reference/services/ecs.html#ECS.Client.run_task | | --- | --- | | 類型： | 覆蓋：dict | | --- | --- | | 類型： | launch_type：str | | --- | --- | ### AWS Batch Service * [AWSBatchOperator](28) ：在AWS Batch Service上執行任務。 #### AWSBatchOperator ``` class airflow.contrib.operators.awsbatch_operator.AWSBatchOperator(job_name, job_definition, job_queue, overrides, max_retries=4200, aws_conn_id=None, region_name=None, **kwargs) ``` 基礎： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 在AWS Batch Service上執行作業 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **job_name** （ _str_ ） - 將在AWS Batch上運行的作業的名稱 * **job_definition** （ _str_ ） - AWS Batch上的作業定義名稱 * **job_queue** （ _str_ ） - AWS Batch上的隊列名稱 * **max_retries** （ _int_ ） - 服務器未合并時的指數退避重試，4200 = 48小時 * **aws_conn_id** （ _str_ ） - AWS憑證/區域名稱的連接ID。如果為None，將使用憑證boto3策略（ [http://boto3.readthedocs.io/en/latest/guide/configuration.html](http://boto3.readthedocs.io/en/latest/guide/configuration.html) ）。 * **region_name** - 要在AWS Hook中使用的區域名稱。覆蓋連接中的region_name（如果提供） | | --- | --- | | 帕拉姆： | 覆蓋：boto3將在containerOverrides上接收的相同參數（模板化）： [http](http://boto3.readthedocs.io/en/latest/reference/services/batch.html) ：//boto3.readthedocs.io/en/latest/reference/services/batch.html#submit_job | | --- | --- | | 類型： | 覆蓋：dict | | --- | --- | ### AWS RedShift * [AwsRedshiftClusterSensor](28) ：等待Redshift群集達到特定狀態。 * [RedshiftHook](28) ：使用boto3庫與AWS Redshift交互。 * [RedshiftToS3Transfer](28) ：對帶有或不帶標頭的CSV執行卸載命令。 * [S3ToRedshiftTransfer](28) ：從S3執行復制命令為CSV，帶或不帶標題。 #### AwsRedshiftClusterSensor ``` class airflow.contrib.sensors.aws_redshift_cluster_sensor.AwsRedshiftClusterSensor(cluster_identifier, target_status='available', aws_conn_id='aws_default', *args, **kwargs) ``` 基礎： [`airflow.sensors.base_sensor_operator.BaseSensorOperator`](code.html "airflow.sensors.base_sensor_operator.BaseSensorOperator") 等待Redshift群集達到特定狀態。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **cluster_identifier** （ _str_ ） - 要ping的集群的標識符。 * **target_status** （ _str_ ） - 所需的集群狀態。 | | --- | --- | ``` poke(context) ``` 傳感器在派生此類時定義的功能應該覆蓋。 #### RedshiftHook ``` class airflow.contrib.hooks.redshift_hook.RedshiftHook(aws_conn_id='aws_default') ``` 基地： [`airflow.contrib.hooks.aws_hook.AwsHook`](code.html "airflow.contrib.hooks.aws_hook.AwsHook") 使用boto3庫與AWS Redshift交互 ``` cluster_status(cluster_identifier) ``` 返回群集的狀態 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **cluster_identifier** （ _str_ ） - 集群的唯一標識符 | | --- | --- | ``` create_cluster_snapshot(snapshot_identifier, cluster_identifier) ``` 創建群集的快照 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **snapshot_identifier** （ _str_ ） - 群集快照的唯一標識符 * **cluster_identifier** （ _str_ ） - 集群的唯一標識符 | | --- | --- | ``` delete_cluster(cluster_identifier, skip_final_cluster_snapshot=True, final_cluster_snapshot_identifier='') ``` 刪除群集并可選擇創建快照 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **cluster_identifier** （ _str_ ） - 集群的唯一標識符 * **skip_final_cluster_snapshot** （ _bool_ ） - 確定群集快照創建 * **final_cluster_snapshot_identifier** （ _str_ ） - 最終集群快照的名稱 | | --- | --- | ``` describe_cluster_snapshots(cluster_identifier) ``` 獲取群集的快照列表 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **cluster_identifier** （ _str_ ） - 集群的唯一標識符 | | --- | --- | ``` restore_from_cluster_snapshot(cluster_identifier, snapshot_identifier) ``` 從其快照還原群集 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **cluster_identifier** （ _str_ ） - 集群的唯一標識符 * **snapshot_identifier**（_str_） - 群集快照的唯一標識符 | | --- | --- | #### RedshiftToS3Transfer ``` class airflow.operators.redshift_to_s3_operator.RedshiftToS3Transfer（schema，table，s3_bucket，s3_key，redshift_conn_id ='redshift_default'，aws_conn_id ='aws_default'，unload_options =（），autocommit = False，parameters = None，include_header = False，* args，* * kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 執行UNLOAD命令，將s3作為帶標題的CSV <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **schema**（_字符串_） - 對redshift數據庫中特定模式的引用 * **table**（_字符串_） - 對redshift數據庫中特定表的引用 * **s3_bucket**（_string_） - 對特定S3存儲桶的引用 * **s3_key**（_字符串_） - 對特定S3密鑰的引用 * **redshift_conn_id**（_string_） - 對特定redshift數據庫的引用 * **aws_conn_id**（_string_） - 對特定S3連接的引用 * **unload_options**（_list_） - 對UNLOAD選項列表的引用 | | --- | --- | #### S3ToRedshiftTransfer ``` class airflow.operators.s3_to_redshift_operator.S3ToRedshiftTransfer（schema，table，s3_bucket，s3_key，redshift_conn_id ='redshift_default'，aws_conn_id ='aws_default'，copy_options =（），autocommit = False，parameters = None，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 執行COPY命令將文件從s3加載到Redshift <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **schema**（_字符串_） - 對redshift數據庫中特定模式的引用 * **table**（_字符串_） - 對redshift數據庫中特定表的引用 * **s3_bucket**（_string_） - 對特定S3存儲桶的引用 * **s3_key**（_字符串_） - 對特定S3密鑰的引用 * **redshift_conn_id**（_string_） - 對特定redshift數據庫的引用 * **aws_conn_id**（_string_） - 對特定S3連接的引用 * **copy_options**（_list_） - 對COPY選項列表的引用 | | --- | --- | ## Databricks [Databricks](https://databricks.com/)貢獻了一個Airflow運算符，可以將運行提交到Databricks平臺。在運營商內部與`api/2.0/jobs/runs/submit` [端點進行通信](https://docs.databricks.com/api/latest/jobs.html)。 ### DatabricksSubmitRunOperator ``` class airflow.contrib.operators.databricks_operator.DatabricksSubmitRunOperator（json = None，spark_jar_task = None，notebook_task = None，new_cluster = None，existing_cluster_id = None，libraries = None，run_name = None，timeout_seconds = None，databricks_conn_id ='databricks_default'，polling_period_seconds = 30，databricks_retry_limit = 3，do_xcom_push = False，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 使用[api / 2.0 / jobs / runs / submit](https://docs.databricks.com/api/latest/jobs.html) API端點向Databricks提交Spark作業運行。 有兩種方法可以實例化此運算符。 在第一種方式，你可以把你通常用它來調用的JSON有效載荷`api/2.0/jobs/runs/submit`端點并將其直接傳遞到我們`DatabricksSubmitRunOperator`通過`json`參數。例如 ``` json = { 'new_cluster' : { 'spark_version' : '2.1.0-db3-scala2.11' , 'num_workers' : 2 }, 'notebook_task' : { 'notebook_path' : '/Users/airflow@example.com/PrepareData' , }, } notebook_run = DatabricksSubmitRunOperator ( task_id = 'notebook_run' , json = json ) ``` 另一種完成同樣事情的方法是直接使用命名參數`DatabricksSubmitRunOperator`。請注意，`runs/submit`端點中的每個頂級參數都只有一個命名參數。在此方法中，您的代碼如下所示： ``` new_cluster = { 'spark_version' : '2.1.0-db3-scala2.11' , 'num_workers' : 2 } notebook_task = { 'notebook_path' : '/Users/airflow@example.com/PrepareData' , } notebook_run = DatabricksSubmitRunOperator ( task_id = 'notebook_run' , new_cluster = new_cluster , notebook_task = notebook_task ) ``` 在提供json參數**和**命名參數的情況下，它們將合并在一起。如果在合并期間存在沖突，則命名參數將優先并覆蓋頂級`json`鍵。 ``` 目前DatabricksSubmitRunOperator支持的命名參數是 ``` * `spark_jar_task` * `notebook_task` * `new_cluster` * `existing_cluster_id` * `libraries` * `run_name` * `timeout_seconds` <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **json**（_dict_） - 包含API參數的JSON對象，將直接傳遞給`api/2.0/jobs/runs/submit`端點。其他命名參數（即`spark_jar_task`，`notebook_task`..）到該運營商將與此JSON字典合并如果提供他們。如果在合并期間存在沖突，則命名參數將優先并覆蓋頂級json鍵。（模板） 也可以看看 有關模板的更多信息，請參閱[Jinja模板](concepts.html)。[https://docs.databricks.com/api/latest/jobs.html#runs-submit](https://docs.databricks.com/api/latest/jobs.html) * **spark_jar_task**（_dict_） - JAR任務的主要類和參數。請注意，實際的JAR在`libraries`。中指定。_無論是_ `spark_jar_task` _或_ `notebook_task`應符合規定。該字段將被模板化。 也可以看看 [https://docs.databricks.com/api/latest/jobs.html#jobssparkjartask](https://docs.databricks.com/api/latest/jobs.html) * **notebook_task**（_dict_） - 筆記本任務的筆記本路徑和參數。_無論是_ `spark_jar_task` _或_ `notebook_task`應符合規定。該字段將被模板化。 也可以看看 [https://docs.databricks.com/api/latest/jobs.html#jobsnotebooktask](https://docs.databricks.com/api/latest/jobs.html) * **new_cluster**（_dict_） - 將在其上運行此任務的新群集的規范。_無論是_ `new_cluster` _或_ `existing_cluster_id`應符合規定。該字段將被模板化。 也可以看看 [https://docs.databricks.com/api/latest/jobs.html#jobsclusterspecnewcluster](https://docs.databricks.com/api/latest/jobs.html) * **existing_cluster_id**（_字符串_） - 要運行此任務的現有集群的ID。_無論是_ `new_cluster` _或_ `existing_cluster_id`應符合規定。該字段將被模板化。 * **圖書館**（_dicts列表_） - 這個運行的庫將使用。該字段將被模板化。 也可以看看 [https://docs.databricks.com/api/latest/libraries.html#managedlibrarieslibrary](https://docs.databricks.com/api/latest/libraries.html) * **run_name**（_string_） - 用于此任務的運行名稱。默認情況下，這將設置為Airflow `task_id`。這`task_id`是超類的必需參數`BaseOperator`。該字段將被模板化。 * **timeout_seconds**（_int32_） - 此次運行的超時。默認情況下，使用值0表示沒有超時。該字段將被模板化。 * **databricks_conn_id**（_string_） - 要使用的Airflow連接的名稱。默認情況下，在常見情況下，這將是`databricks_default`。要使用基于令牌的身份驗證，請`token`在連接的額外字段中提供密鑰。 * **polling_period_seconds**（_int_） - 控制我們輪詢此運行結果的速率。默認情況下，操作員每30秒輪詢一次。 * **databricks_retry_limit**（_int_） - 如果Databricks后端無法訪問，則重試的次數。其值必須大于或等于1。 * **do_xcom_push**（_boolean_） - 我們是否應該將run_id和run_page_url推送到xcom。 | | --- | --- | ## GCP：Google云端平臺 Airflow廣泛支持Google Cloud Platform。但請注意，大多數Hooks和Operators都在contrib部分。這意味著他們具有_beta_狀態，這意味著他們可以在次要版本之間進行重大更改。 請參閱[GCP連接類型](howto/manage-connections.html)文檔以配置與GCP的連接。 ### 記錄 可以將Airflow配置為在Google云端存儲中讀取和寫入任務日志。請參閱[將日志寫入Google云端存儲](howto/write-logs.html)。 ### BigQuery的 #### BigQuery運算符 * [BigQueryCheckOperator](28)：對SQL查詢執行檢查，該查詢將返回具有不同值的單行。 * [BigQueryValueCheckOperator](28)：使用SQL代碼執行簡單的值檢查。 * [BigQueryIntervalCheckOperator](28)：檢查作為SQL表達式給出的度量值是否在days_back之前的某個容差范圍內。 * [BigQueryCreateEmptyTableOperator](28)：在指定的BigQuery數據集中創建一個新的空表，可選擇使用模式。 * [BigQueryCreateExternalTableOperator](28)：使用Google Cloud Storage中的數據在數據集中創建新的外部表。 * [BigQueryDeleteDatasetOperator](28)：刪除現有的BigQuery數據集。 * [BigQueryOperator](28)：在特定的BigQuery數據庫中執行BigQuery SQL查詢。 * [BigQueryToBigQueryOperator](28)：將BigQuery表復制到另一個BigQuery表。 * [BigQueryToCloudStorageOperator](28)：將BigQuery表傳輸到Google Cloud Storage存儲桶 ##### BigQueryCheckOperator ``` class airflow.contrib.operators.bigquery_check_operator.BigQueryCheckOperator（sql，bigquery_conn_id ='bigquery_default'，* args，** kwargs） ``` 基地： [`airflow.operators.check_operator.CheckOperator`](code.html "airflow.operators.check_operator.CheckOperator") 對BigQuery執行檢查。該`BigQueryCheckOperator`預期的SQL查詢將返回一行。使用python `bool`強制轉換評估第一行的每個值。如果任何值返回，`False`則檢查失敗并輸出錯誤。 請注意，Python bool強制轉換如下`False`： * `False` * `0` * 空字符串（`""`） * 空列表（`[]`） * 空字典或集（`{}`） 給定一個查詢，它只會在計數時失敗。您可以制作更復雜的查詢，例如，可以檢查表與上游源表的行數相同，或者今天的分區計數大于昨天的分區，或者一組指標是否更少7天平均值超過3個標準差。`SELECT COUNT(*) FROM foo``== 0` 此運算符可用作管道中的數據質量檢查，并且根據您在DAG中的位置，您可以選擇停止關鍵路徑，防止發布可疑數據，或者在旁邊接收電子郵件替代品阻止DAG的進展。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **sql**（_string_） - 要執行的sql * **bigquery_conn_id**（_string_） - 對BigQuery數據庫的引用 | | --- | --- | ##### BigQueryValueCheckOperator ``` class airflow.contrib.operators.bigquery_check_operator.BigQueryValueCheckOperator（sql，pass_value，tolerance = None，bigquery_conn_id ='bigquery_default'，* args，** kwargs） ``` 基地： [`airflow.operators.check_operator.ValueCheckOperator`](code.html "airflow.operators.check_operator.ValueCheckOperator") 使用sql代碼執行簡單的值檢查。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **sql**（_string_） - 要執行的sql | | --- | --- | ##### BigQueryIntervalCheckOperator ``` class airflow.contrib.operators.bigquery_check_operator.BigQueryIntervalCheckOperator（table，metrics_thresholds，date_filter_column ='ds'，days_back = -7，bigquery_conn_id ='bigquery_default'，* args，** kwargs） ``` 基地： [`airflow.operators.check_operator.IntervalCheckOperator`](code.html "airflow.operators.check_operator.IntervalCheckOperator") 檢查作為SQL表達式給出的度量值是否在days_back之前的某個容差范圍內。 此方法構造一個類似的查詢 ``` SELECT { metrics_threshold_dict_key } FROM { table } WHERE { date_filter_column } =< date > ``` <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **table**（_str_） - 表名 * **days_back**（_int_） - ds與我們要檢查的ds之間的天數。默認為7天 * **metrics_threshold**（_dict_） - 由指標索引的比率字典，例如'COUNT（*）'：1.5將需要當前日和之前的days_back之間50％或更小的差異。 | | --- | --- | ##### BigQueryGetDataOperator ``` class airflow.contrib.operators.bigquery_get_data.BigQueryGetDataOperator（dataset_id，table_id，max_results ='100'，selected_fields = None，bigquery_conn_id ='bigquery_default'，delegate_to = None，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 從BigQuery表中獲取數據（或者為所選列獲取數據）并在python列表中返回數據。返回列表中的元素數將等于獲取的行數。列表中的每個元素將再次是一個列表，其中元素將表示該行的列值。 **結果示例**：`[['Tony', '10'], ['Mike', '20'], ['Steve', '15']]` 注意 如果傳遞的字段`selected_fields`的順序與BQ表中已有的列的順序不同，則數據仍將按BQ表的順序排列。例如，如果BQ表有3列，`[A,B,C]`并且您傳遞'B，`selected_fields`那么數據中的A' 仍然是表格`'A,B'`。 **示例** ： ``` get_data = BigQueryGetDataOperator ( task_id = 'get_data_from_bq' , dataset_id = 'test_dataset' , table_id = 'Transaction_partitions' , max_results = '100' , selected_fields = 'DATE' , bigquery_conn_id = 'airflow-service-account' ) ``` <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **dataset_id** - 請求的表的數據集ID。（模板） * **table_id**（_string_） - 請求表的表ID。（模板） * **max_results**（_string_） - 從表中獲取的最大記錄數（行數）。（模板） * **selected_fields**（_string_） - 要返回的字段列表（逗號分隔）。如果未指定，則返回所有字段。 * **bigquery_conn_id**（_string_） - 對特定BigQuery鉤子的引用。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 | | --- | --- | ##### BigQueryCreateEmptyTableOperator ``` class airflow.contrib.operators.bigquery_operator.BigQueryCreateEmptyTableOperator（dataset_id，table_id，project_id = None，schema_fields = None，gcs_schema_object = None，time_partitioning = {}，bigquery_conn_id ='bigquery_default'，google_cloud_storage_conn_id ='google_cloud_default'，delegate_to = None，* args ，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 在指定的BigQuery數據集中創建一個新的空表，可選擇使用模式。 可以用兩種方法之一指定用于BigQuery表的模式。您可以直接傳遞架構字段，也可以將運營商指向Google云存儲對象名稱。Google云存儲中的對象必須是包含架構字段的JSON文件。您還可以創建沒有架構的表。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **project_id**（_string_） - 將表創建的項目。（模板） * **dataset_id**（_string_） - 用于創建表的數據集。（模板） * **table_id**（_string_） - 要創建的表的名稱。（模板） * **schema_fields**（_list_） - 如果設置，則此處定義的架構字段列表：[https](https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs)：[//cloud.google.com/bigquery/docs/reference/rest/v2/jobs#configuration.load.schema](https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs) **示例** ： ``` schema_fields = [{ "name" : "emp_name" , "type" : "STRING" , "mode" : "REQUIRED" }, { "name" : "salary" , "type" : "INTEGER" , "mode" : "NULLABLE" }] ``` * **gcs_schema_object**（_string_） - 包含模式（模板化）的JSON文件的完整路徑。例如：`gs://test-bucket/dir1/dir2/employee_schema.json` * **time_partitioning**（_dict_） - 配置可選的時間分區字段，即按API規范按字段，類型和到期分區。 也可以看看 [https://cloud.google.com/bigquery/docs/reference/rest/v2/tables#timePartitioning](https://cloud.google.com/bigquery/docs/reference/rest/v2/tables) * **bigquery_conn_id**（_string_） - 對特定BigQuery掛鉤的引用。 * **google_cloud_storage_conn_id**（_字符串_） - 對特定Google云存儲掛鉤的引用。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 | | --- | --- | **示例（在GCS中使用模式JSON）**： ``` CreateTable = BigQueryCreateEmptyTableOperator ( task_id = 'BigQueryCreateEmptyTableOperator_task' , dataset_id = 'ODS' , table_id = 'Employees' , project_id = 'internal-gcp-project' , gcs_schema_object = 'gs://schema-bucket/employee_schema.json' , bigquery_conn_id = 'airflow-service-account' , google_cloud_storage_conn_id = 'airflow-service-account' ) ``` **對應的Schema文件**（`employee_schema.json`）： ``` [ { "mode" : "NULLABLE" , "name" : "emp_name" , "type" : "STRING" }, { "mode" : "REQUIRED" , "name" : "salary" , "type" : "INTEGER" } ] ``` **示例（在DAG中使用模式）**： ``` CreateTable = BigQueryCreateEmptyTableOperator ( task_id = 'BigQueryCreateEmptyTableOperator_task' , dataset_id = 'ODS' , table_id = 'Employees' , project_id = 'internal-gcp-project' , schema_fields = [{ "name" : "emp_name" , "type" : "STRING" , "mode" : "REQUIRED" }, { "name" : "salary" , "type" : "INTEGER" , "mode" : "NULLABLE" }], bigquery_conn_id = 'airflow-service-account' , google_cloud_storage_conn_id = 'airflow-service-account' ) ``` ##### BigQueryCreateExternalTableOperator ``` class airflow.contrib.operators.bigquery_operator.BigQueryCreateExternalTableOperator（bucket，source_objects，destination_project_dataset_table，schema_fields = None，schema_object = None，source_format ='CSV'，compression ='NONE'，skip_leading_rows = 0，field_delimiter ='，'，max_bad_records = 0 ，quote_character = None，allow_quoted_newlines = False，allow_jagged_rows = False，bigquery_conn_id ='bigquery_default'，google_cloud_storage_conn_id ='google_cloud_default'，delegate_to = None，src_fmt_configs = {}，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 使用Google云端存儲中的數據在數據集中創建新的外部表。 可以用兩種方法之一指定用于BigQuery表的模式。您可以直接傳遞架構字段，也可以將運營商指向Google云存儲對象名稱。Google云存儲中的對象必須是包含架構字段的JSON文件。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_string_） - 指向外部表的存儲桶。（模板） * **source_objects** - 指向表格的Google云存儲URI列表。（模板化）如果source_format是'DATASTORE_BACKUP'，則列表必須只包含一個URI。 * **destination_project_dataset_table**（_string_） - 用于將數據加載到（模板化）的虛線（<project>。）<dataset>。<table> BigQuery表。如果未包含<project>，則項目將是連接json中定義的項目。 * **schema_fields**（_list_） - 如果設置，則此處定義的架構字段列表：[https](https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs)：[//cloud.google.com/bigquery/docs/reference/rest/v2/jobs#configuration.load.schema](https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs) **示例** ： ``` schema_fields = [{ "name" : "emp_name" , "type" : "STRING" , "mode" : "REQUIRED" }, { "name" : "salary" , "type" : "INTEGER" , "mode" : "NULLABLE" }] ``` 當source_format為'DATASTORE_BACKUP'時，不應設置。 * **schema_object** - 如果設置，則指向包含表的架構的.json文件的GCS對象路徑。（模板） * **schema_object** - 字符串 * **source_format**（_string_） - 數據的文件格式。 * **compression**（_字符串_） - [可選]數據源的壓縮類型。可能的值包括GZIP和NONE。默認值為NONE。Google Cloud Bigtable，Google Cloud Datastore備份和Avro格式會忽略此設置。 * **skip_leading_rows**（_int_） - 從CSV加載時要跳過的行數。 * **field_delimiter**（_string_） - 用于CSV的分隔符。 * **max_bad_records**（_int_） - BigQuery在運行作業時可以忽略的最大錯誤記錄數。 * **quote_character**（_string_） - 用于引用CSV文件中數據部分的值。 * **allow_quoted_newlines**（_boolean_） - 是否允許引用的換行符（true）或不允許（false）。 * **allow_jagged_rows**（_bool_） - 接受缺少尾隨可選列的行。缺失值被視為空值。如果為false，則缺少尾隨列的記錄將被視為錯誤記錄，如果錯誤記錄太多，則會在作業結果中返回無效錯誤。僅適用于CSV，忽略其他格式。 * **bigquery_conn_id**（_string_） - 對特定BigQuery掛鉤的引用。 * **google_cloud_storage_conn_id**（_字符串_） - 對特定Google云存儲掛鉤的引用。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **src_fmt_configs**（_dict_） - 配置特定于源格式的可選字段 | | --- | --- | ##### BigQueryDeleteDatasetOperator ##### BigQueryOperator ``` class airflow.contrib.operators.bigquery_operator.BigQueryOperator（bql = None，sql = None，destination_dataset_table = False，write_disposition ='WRITE_EMPTY'，allow_large_results = False，flatten_results = False，bigquery_conn_id ='bigquery_default'，delegate_to = None，udf_config = False ，use_legacy_sql = True，maximum_billing_tier = None，maximum_bytes_billed = None，create_disposition ='CREATE_IF_NEEDED'，schema_update_options =（），query_params = None，priority ='INTERACTIVE'，time_partitioning = {}，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 在特定的BigQuery數據庫中執行BigQuery SQL查詢 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **BQL**（_可接收表示SQL語句中的海峽_ _，_ _海峽列表_ _（_ _SQL語句_ _）_ _，或_ _參照模板文件模板引用在“.SQL”結束海峽認可。_） - （不推薦使用。<cite>SQL</cite>參數代替）要執行的sql代碼（模板化） * **SQL**（_可接收表示SQL語句中的海峽_ _，_ _海峽列表_ _（_ _SQL語句_ _）_ _，或_ _參照模板文件模板引用在“.SQL”結束海峽認可。_） - SQL代碼被執行（模板） * **destination_dataset_table**（_string_） - 一個虛線（<project>。| <project>：）<dataset>。<table>，如果設置，將存儲查詢結果。（模板） * **write_disposition**（_string_） - 指定目標表已存在時發生的操作。（默認：'WRITE_EMPTY'） * **create_disposition**（_string_） - 指定是否允許作業創建新表。（默認值：'CREATE_IF_NEEDED'） * **allow_large_results**（_boolean_） - 是否允許大結果。 * **flatten_results**（_boolean_） - 如果為true且查詢使用舊版SQL方言，則展平查詢結果中的所有嵌套和重復字段。`allow_large_results`必須是`true`如果設置為`false`。對于標準SQL查詢，將忽略此標志，并且結果永遠不會展平。 * **bigquery_conn_id**（_string_） - 對特定BigQuery鉤子的引用。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **udf_config**（_list_） - 查詢的用戶定義函數配置。有關詳細信息，請參閱[https://cloud.google.com/bigquery/user-defined-functions](https://cloud.google.com/bigquery/user-defined-functions)。 * **use_legacy_sql**（_boolean_） - 是使用舊SQL（true）還是標準SQL（false）。 * **maximum_billing_tier**（_整數_） - 用作基本價格乘數的正整數。默認為None，在這種情況下，它使用項目中設置的值。 * **maximum_bytes_billed**（_float_） - 限制為此作業計費的字節數。超出此限制的字節數的查詢將失敗（不會產生費用）。如果未指定，則將其設置為項目默認值。 * **schema_update_options**（_tuple_） - 允許更新目標表的模式作為加載作業的副作用。 * **query_params**（_dict_） - 包含查詢參數類型和值的字典，傳遞給BigQuery。 * **priority**（_字符串_） - 指定查詢的優先級。可能的值包括INTERACTIVE和BATCH。默認值為INTERACTIVE。 * **time_partitioning**（_dict_） - 配置可選的時間分區字段，即按API規范按字段，類型和到期分區。請注意，'field'不能與dataset.table $ partition一起使用。 | | --- | --- | ##### BigQueryTableDeleteOperator ``` class airflow.contrib.operators.bigquery_table_delete_operator.BigQueryTableDeleteOperator（deletion_dataset_table，bigquery_conn_id ='bigquery_default'，delegate_to = None，ignore_if_missing = False，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 刪除BigQuery表 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **deletion_dataset_table**（_string_） - 一個虛線（<project>。| <project>：）<dataset>。<table>，指示將刪除哪個表。（模板） * **bigquery_conn_id**（_string_） - 對特定BigQuery鉤子的引用。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **ignore_if_missing**（_boolean_） - 如果為True，則即使請求的表不存在也返回成功。 | | --- | --- | ##### BigQueryToBigQueryOperator ``` class airflow.contrib.operators.bigquery_to_bigquery.BigQueryToBigQueryOperator（source_project_dataset_tables，destination_project_dataset_table，write_disposition ='WRITE_EMPTY'，create_disposition ='CREATE_IF_NEEDED'，bigquery_conn_id ='bigquery_default'，delegate_to = None，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 將數據從一個BigQuery表復制到另一個。 也可以看看 有關這些參數的詳細信息，請訪問：[https](https://cloud.google.com/bigquery/docs/reference/v2/jobs)：[//cloud.google.com/bigquery/docs/reference/v2/jobs#configuration.copy](https://cloud.google.com/bigquery/docs/reference/v2/jobs) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **source_project_dataset_tables**（_list | string_） - 一個或多個點（項目：[|](28)項目。）<dataset>。<table>用作源數據的BigQuery表。如果未包含<project>，則項目將是連接json中定義的項目。如果有多個源表，請使用列表。（模板） * **destination_project_dataset_table**（_string_） - 目標BigQuery表。格式為：（project：[|](28) project。）<dataset>。<table>（模板化） * **write_disposition**（_string_） - 表已存在時的寫處置。 * **create_disposition**（_string_） - 如果表不存在，則創建處置。 * **bigquery_conn_id**（_string_） - 對特定BigQuery鉤子的引用。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 | | --- | --- | ##### BigQueryToCloudStorageOperator ``` class airflow.contrib.operators.bigquery_to_gcs.BigQueryToCloudStorageOperator（source_project_dataset_table，destination_cloud_storage_uris，compression ='NONE'，export_format ='CSV'，field_delimiter ='，'，print_header = True，bigquery_conn_id ='bigquery_default'，delegate_to = None，* args， ** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 將BigQuery表傳輸到Google Cloud Storage存儲桶。 也可以看看 有關這些參數的詳細信息，請訪問：[https](https://cloud.google.com/bigquery/docs/reference/v2/jobs)：[//cloud.google.com/bigquery/docs/reference/v2/jobs](https://cloud.google.com/bigquery/docs/reference/v2/jobs) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **source_project_dataset_table**（_string_） - 用作源數據的虛線（<project>。| <project>：）<dataset>。<table> BigQuery表。如果未包含<project>，則項目將是連接json中定義的項目。（模板） * **destination_cloud_storage_uris**（_list_） - 目標Google云端存儲URI（例如gs：//some-bucket/some-file.txt）。（模板化）遵循此處定義的慣例：https：//cloud.google.com/bigquery/exporting-data-from-bigquery#exportingmultiple * **compression**（_string_） - 要使用的壓縮類型。 * **export_format** - 要導出的文件格式。 * **field_delimiter**（_string_） - 提取到CSV時使用的分隔符。 * **print_header**（_boolean_） - 是否打印CSV文件提取的標頭。 * **bigquery_conn_id**（_string_） - 對特定BigQuery鉤子的引用。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 | | --- | --- | #### BigQueryHook ``` class airflow.contrib.hooks.bigquery_hook.BigQueryHook（bigquery_conn_id ='bigquery_default'，delegate_to = None，use_legacy_sql = True） ``` 基地：[`airflow.contrib.hooks.gcp_api_base_hook.GoogleCloudBaseHook`](code.html "airflow.contrib.hooks.gcp_api_base_hook.GoogleCloudBaseHook")，[`airflow.hooks.dbapi_hook.DbApiHook`](code.html "airflow.hooks.dbapi_hook.DbApiHook")，`airflow.utils.log.logging_mixin.LoggingMixin` 與BigQuery交互。此掛鉤使用Google Cloud Platform連接。 ``` get_conn（） ``` 返回BigQuery PEP 249連接對象。 ``` get_pandas_df（sql，parameters = None，dialect = None） ``` 返回BigQuery查詢生成的結果的Pandas DataFrame。必須重寫DbApiHook方法，因為Pandas不支持PEP 249連接，但SQLite除外。看到： [https://github.com/pydata/pandas/blob/master/pandas/io/sql.py#L447 ](https://github.com/pydata/pandas/blob/master/pandas/io/sql.py)[https://github.com/pydata/pandas/issues/6900](https://github.com/pydata/pandas/issues/6900) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **sql**（_string_） - 要執行的BigQuery SQL。 * **參數**（_映射_ _或_ _可迭代_） - 用于呈現SQL查詢的參數（未使用，請保留覆蓋超類方法） * **dialect**（_{'legacy' __，_ _'standard'}中的__字符串_） - BigQuery SQL的方言 - 遺留SQL或標準SQL默認使用<cite>self.use_legacy_sql（</cite>如果未指定） | | --- | --- | ``` get_service（） ``` 返回一個BigQuery服務對象。 ``` insert_rows（table，rows，target_fields = None，commit_every = 1000） ``` 目前不支持插入。從理論上講，您可以使用BigQuery的流API將行插入表中，但這尚未實現。 ``` table_exists（project_id，dataset_id，table_id） ``` 檢查Google BigQuery中是否存在表格。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **project_id**（_字符串_） - 要在其中查找表的Google云項目。提供給鉤子的連接必須提供對指定項目的訪問。 * **dataset_id**（_string_） - 要在其中查找表的數據集的名稱。 * **table_id**（_string_） - 要檢查的表的名稱。 | | --- | --- | ### 云DataFlow #### DataFlow運算符 * [DataFlowJavaOperator](28)：啟動用Java編寫的Cloud Dataflow作業。 * [DataflowTemplateOperator](28)：啟動模板化的Cloud DataFlow批處理作業。 * [DataFlowPythonOperator](28)：啟動用python編寫的Cloud Dataflow作業。 ##### DataFlowJavaOperator ``` class airflow.contrib.operators.dataflow_operator.DataFlowJavaOperator（jar，dataflow_default_options = None，options = None，gcp_conn_id ='google_cloud_default'，delegate_to = None，poll_sleep = 10，job_class = None，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 啟動Java Cloud DataFlow批處理作業。操作的參數將傳遞給作業。 在dag的default_args中定義dataflow_ *參數是一個很好的做法，例如項目，區域和分段位置。 ``` default_args = { 'dataflow_default_options' : { 'project' : 'my-gcp-project' , 'zone' : 'europe-west1-d' , 'stagingLocation' : 'gs://my-staging-bucket/staging/' } } ``` 您需要使用`jar`參數將路徑作為文件引用傳遞給數據流，jar需要是一個自動執行的jar（請參閱以下文檔：[https](https://beam.apache.org/documentation/runners/dataflow/)：[//beam.apache.org/documentation/runners/dataflow/#self-執行jar](https://beam.apache.org/documentation/runners/dataflow/)。使用`options`轉嫁選項你的工作。 ``` t1 = DataFlowOperation ( task_id = 'datapflow_example' , jar = '{{var.value.gcp_dataflow_base}}pipeline/build/libs/pipeline-example-1.0.jar' , options = { 'autoscalingAlgorithm' : 'BASIC' , 'maxNumWorkers' : '50' , 'start' : '{{ds}}' , 'partitionType' : 'DAY' , 'labels' : { 'foo' : 'bar' } }, gcp_conn_id = 'gcp-airflow-service-account' , dag = my - dag ) ``` 這兩個`jar`和`options`模板化，所以你可以在其中使用變量。 ``` default_args = { 'owner' : 'airflow' , 'depends_on_past' : False , 'start_date' : ( 2016 , 8 , 1 ), 'email' : [ 'alex@vanboxel.be' ], 'email_on_failure' : False , 'email_on_retry' : False , 'retries' : 1 , 'retry_delay' : timedelta ( minutes = 30 ), 'dataflow_default_options' : { 'project' : 'my-gcp-project' , 'zone' : 'us-central1-f' , 'stagingLocation' : 'gs://bucket/tmp/dataflow/staging/' , } } dag = DAG ( 'test-dag' , default_args = default_args ) task = DataFlowJavaOperator ( gcp_conn_id = 'gcp_default' , task_id = 'normalize-cal' , jar = '{{var.value.gcp_dataflow_base}}pipeline-ingress-cal-normalize-1.0.jar' , options = { 'autoscalingAlgorithm' : 'BASIC' , 'maxNumWorkers' : '50' , 'start' : '{{ds}}' , 'partitionType' : 'DAY' }, dag = dag ) ``` ##### DataflowTemplateOperator ``` class airflow.contrib.operators.dataflow_operator.DataflowTemplateOperator（template，dataflow_default_options = None，parameters = None，gcp_conn_id ='google_cloud_default'，delegate_to = None，poll_sleep = 10，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 啟動模板化云DataFlow批處理作業。操作的參數將傳遞給作業。在dag的default_args中定義dataflow_ *參數是一個很好的做法，例如項目，區域和分段位置。 也可以看看 [https://cloud.google.com/dataflow/docs/reference/rest/v1b3/LaunchTemplateParameters ](https://cloud.google.com/dataflow/docs/reference/rest/v1b3/LaunchTemplateParameters)[https://cloud.google.com/dataflow/docs/reference/rest/v1b3/RuntimeEnvironment](https://cloud.google.com/dataflow/docs/reference/rest/v1b3/RuntimeEnvironment) ``` default_args = { 'dataflow_default_options' : { 'project' : 'my-gcp-project' 'zone' : 'europe-west1-d' , 'tempLocation' : 'gs://my-staging-bucket/staging/' } } } ``` 您需要將路徑作為帶`template`參數的文件引用傳遞給數據流模板。使用`parameters`來傳遞參數給你的工作。使用`environment`對運行環境變量傳遞給你的工作。 ``` t1 = DataflowTemplateOperator ( task_id = 'datapflow_example' , template = '{{var.value.gcp_dataflow_base}}' , parameters = { 'inputFile' : "gs://bucket/input/my_input.txt" , 'outputFile' : "gs://bucket/output/my_output.txt" }, gcp_conn_id = 'gcp-airflow-service-account' , dag = my - dag ) ``` `template`，`dataflow_default_options`并且`parameters`是模板化的，因此您可以在其中使用變量。 ##### DataFlowPythonOperator ``` class airflow.contrib.operators.dataflow_operator.DataFlowPythonOperator（py_file，py_options = None，dataflow_default_options = None，options = None，gcp_conn_id ='google_cloud_default'，delegate_to = None，poll_sleep = 10，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") ``` 執行（上下文） ``` 執行python數據流作業。 #### DataFlowHook ``` class airflow.contrib.hooks.gcp_dataflow_hook.DataFlowHook（gcp_conn_id ='google_cloud_default'，delegate_to = None，poll_sleep = 10） ``` 基地： [`airflow.contrib.hooks.gcp_api_base_hook.GoogleCloudBaseHook`](code.html "airflow.contrib.hooks.gcp_api_base_hook.GoogleCloudBaseHook") ``` get_conn（） ``` 返回Google云端存儲服務對象。 ### Cloud DataProc #### DataProc運算符 * [DataprocClusterCreateOperator](28)：在Google Cloud Dataproc上創建新群集。 * [DataprocClusterDeleteOperator](28)：刪除Google Cloud Dataproc上的群集。 * [DataprocClusterScaleOperator](28)：在Google Cloud Dataproc上向上或向下擴展群集。 * [DataProcPigOperator](28)：在Cloud DataProc集群上啟動Pig查詢作業。 * [DataProcHiveOperator](28)：在Cloud DataProc群集上啟動Hive查詢作業。 * [DataProcSparkSqlOperator](28)：在Cloud DataProc集群上啟動Spark SQL查詢作業。 * [DataProcSparkOperator](28)：在Cloud DataProc集群上啟動Spark作業。 * [DataProcHadoopOperator](28)：在Cloud DataProc集群上啟動Hadoop作業。 * [DataProcPySparkOperator](28)：在Cloud DataProc群集上啟動PySpark作業。 * [DataprocWorkflowTemplateInstantiateOperator](28)：在Google Cloud Dataproc上實例化WorkflowTemplate。 * [DataprocWorkflowTemplateInstantiateInlineOperator](28)：在Google Cloud Dataproc上實例化WorkflowTemplate內聯。 ##### DataprocClusterCreateOperator ``` class airflow.contrib.operators.dataproc_operator.DataprocClusterCreateOperator（cluster_name，project_id，num_workers，zone，network_uri = None，subnetwork_uri = None，internal_ip_only = None，tags = None，storage_bucket = None，init_actions_uris = None，init_action_timeout ='10m'，metadata =無，image_version =無，屬性=無，master_machine_type ='n1-standard-4'，master_disk_size = 500，worker_machine_type ='n1-standard-4'，worker_disk_size = 500，num_preemptible_workers = 0，labels = None，region =' global'，gcp_conn_id ='google_cloud_default'，delegate_to = None，service_account = None，service_account_scopes = None，idle_delete_ttl = None，auto_delete_time = None，auto_delete_ttl = None，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 在Google Cloud Dataproc上創建新群集。操作員將等待創建成功或創建過程中發生錯誤。 參數允許配置群集。請參閱 [https://cloud.google.com/dataproc/docs/reference/rest/v1/projects.regions.clusters](https://cloud.google.com/dataproc/docs/reference/rest/v1/projects.regions.clusters) 有關不同參數的詳細說明。鏈接中詳述的大多數配置參數都可作為此運算符的參數。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **cluster_name**（_string_） - 要創建的DataProc集群的名稱。（模板） * **project_id**（_字符串_） - 用于創建集群的Google云項目的ID。（模板） * **num_workers**（_int_） - 旋轉的工人數量 * **storage_bucket**（_string_） - 要使用的存儲桶，設置為None允許dataproc為您生成自定義存儲桶 * **init_actions_uris**（_list __[ __string __]_） - 包含數據空間初始化腳本的GCS uri列表 * **init_action_timeout**（_string_） - init_actions_uris中可執行腳本必須完成的時間 * **元數據**（_字典_） - 要添加到所有實例的鍵值google計算引擎元數據條目的字典 * **image_version**（_string_） - Dataproc集群內的軟件版本 * **屬性**（_字典_） -性能上的配置文件設置的字典（如火花defaults.conf），見[https://cloud.google.com/dataproc/docs/reference/rest/v1/](https://cloud.google.com/dataproc/docs/reference/rest/v1/) projects.regions.clusters＃SoftwareConfig * **master_machine_type**（_string_） - 計算要用于主節點的引擎機器類型 * **master_disk_size**（_int_） - 主節點的磁盤大小 * **worker_machine_type**（_字符串_） - 計算要用于工作節點的引擎計算機類型 * **worker_disk_size**（_int_） - 工作節點的磁盤大小 * **num_preemptible_workers**（_int_） - 要旋轉的可搶占工作節點數 * **labels**（_dict_） - 要添加到集群的標簽的字典 * **zone**（_字符串_） - 群集所在的區域。（模板） * **network_uri**（_字符串_） - 用于機器通信的網絡uri，不能用subnetwork_uri指定 * **subnetwork_uri**（_字符串_） - 無法使用network_uri指定要用于機器通信的子網uri * **internal_ip_only**（_bool_） - 如果為true，則群集中的所有實例將只具有內部IP地址。這只能為啟用子網的網絡啟用 * **tags**（_list __[ __string __]_） - 要添加到所有實例的GCE標記 * **地區** - 作為'全球'留下，可能在未來變得相關。（模板） * **gcp_conn_id**（_字符串_） - 用于連接到Google Cloud Platform的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **service_account**（_string_） - dataproc實例的服務帳戶。 * **service_account_scopes**（_list __[ __string __]_） - 要包含的服務帳戶范圍的URI。 * **idle_delete_ttl**（_int_） - 群集在保持空閑狀態時保持活動狀態的最長持續時間。通過此閾值將導致群集被自動刪除。持續時間（秒）。 * **auto_delete_time**（_datetime.datetime_） - 自動刪除群集的時間。 * **auto_delete_ttl**（_int_） - 群集的生命周期，群集將在此持續時間結束時自動刪除。持續時間（秒）。（如果設置了auto_delete_time，則將忽略此參數） | | --- | --- | ##### DataprocClusterScaleOperator ``` class airflow.contrib.operators.dataproc_operator.DataprocClusterScaleOperator（cluster_name，project_id，region ='global'，gcp_conn_id ='google_cloud_default'，delegate_to = None，num_workers = 2，num_preemptible_workers = 0，graceful_decommission_timeout = None，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 在Google Cloud Dataproc上進行擴展，向上或向下擴展。操作員將等待，直到重新調整群集。 **示例** ： ``` t1 = DataprocClusterScaleOperator（ ``` task_id ='dataproc_scale'，project_id ='my-project'，cluster_name ='cluster-1'，num_workers = 10，num_preemptible_workers = 10，graceful_decommission_timeout ='1h'dag = dag） 也可以看看 有關擴展群集的更多詳細信息，請參閱以下參考：[https](https://cloud.google.com/dataproc/docs/concepts/configuring-clusters/scaling-clusters)：[//cloud.google.com/dataproc/docs/concepts/configuring-clusters/scaling-clusters](https://cloud.google.com/dataproc/docs/concepts/configuring-clusters/scaling-clusters) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **cluster_name**（_string_） - 要擴展的集群的名稱。（模板） * **project_id**（_字符串_） - 群集運行的Google云項目的ID。（模板） * **region**（_字符串_） - 數據通路簇的區域。（模板） * **gcp_conn_id**（_字符串_） - 用于連接到Google Cloud Platform的連接ID。 * **num_workers**（_int_） - 新的工人數量 * **num_preemptible_workers**（_int_） - 新的可搶占工人數量 * **graceful_decommission_timeout**（_string_） - 優雅的YARN decomissioning超時。最大值為1d * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 | | --- | --- | ##### DataprocClusterDeleteOperator ``` class airflow.contrib.operators.dataproc_operator.DataprocClusterDeleteOperator（cluster_name，project_id，region ='global'，gcp_conn_id ='google_cloud_default'，delegate_to = None，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 刪除Google Cloud Dataproc上的群集。操作員將等待，直到群集被銷毀。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **cluster_name**（_字符串_） - 要創建的集群的名稱。（模板） * **project_id**（_字符串_） - 群集運行的Google云項目的ID。（模板） * **region**（_字符串_） - 保留為“全局”，將來可能會變得相關。（模板） * **gcp_conn_id**（_字符串_） - 用于連接到Google Cloud Platform的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 | | --- | --- | ##### DataProcPigOperator ``` class airflow.contrib.operators.dataproc_operator.DataProcPigOperator（query = None，query_uri = None，variables = None，job_name ='{{task.task_id}} _ {{ds_nodash}}'，cluster_name ='cluster-1'，dataproc_pig_properties =無，dataproc_pig_jars =無，gcp_conn_id ='google_cloud_default'，delegate_to =無，region ='全局'，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 在Cloud DataProc群集上啟動Pig查詢作業。操作的參數將傳遞給集群。 在dag的default_args中定義dataproc_ *參數是一種很好的做法，比如集群名稱和UDF。 ``` default_args = { 'cluster_name' : 'cluster-1' , 'dataproc_pig_jars' : [ 'gs://example/udf/jar/datafu/1.2.0/datafu.jar' , 'gs://example/udf/jar/gpig/1.2/gpig.jar' ] } ``` 您可以將pig腳本作為字符串或文件引用傳遞。使用變量傳遞要在群集上解析的pig腳本的變量，或者使用要在腳本中解析的參數作為模板參數。 **示例** ： ``` t1 = DataProcPigOperator ( task_id = 'dataproc_pig' , query = 'a_pig_script.pig' , variables = { 'out' : 'gs://example/output/{{ds}}' }, dag = dag ) ``` 也可以看看 有關工作提交的更多詳細信息，請參閱以下參考：[https](https://cloud.google.com/dataproc/reference/rest/v1/projects.regions.jobs)：[//cloud.google.com/dataproc/reference/rest/v1/projects.regions.jobs](https://cloud.google.com/dataproc/reference/rest/v1/projects.regions.jobs) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **query**（_string_） - 對查詢文件的查詢或引用（pg或pig擴展）。（模板） * **query_uri**（_string_） - 云存儲上的豬腳本的uri。 * **variables**（_dict_） - 查詢的命名參數的映射。（模板） * **job_name**（_string_） - DataProc集群中使用的作業名稱。默認情況下，此名稱是附加執行數據的task_id，但可以進行模板化。該名稱將始終附加一個隨機數，以避免名稱沖突。（模板） * **cluster_name**（_string_） - DataProc集群的名稱。（模板） * **dataproc_pig_properties**（_dict_） - Pig屬性的映射。非常適合放入默認參數 * **dataproc_pig_jars**（_list_） - 在云存儲中配置的jars的URI（例如：用于UDF和lib），非常適合放入默認參數。 * **gcp_conn_id**（_字符串_） - 用于連接到Google Cloud Platform的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **region**（_字符串_） - 創建數據加載集群的指定區域。 | | --- | --- | ##### DataProcHiveOperator ``` class airflow.contrib.operators.dataproc_operator.DataProcHiveOperator（query = None，query_uri = None，variables = None，job_name ='{{task.task_id}} _ {{ds_nodash}}'，cluster_name ='cluster-1'，dataproc_hive_properties =無，dataproc_hive_jars =無，gcp_conn_id ='google_cloud_default'，delegate_to =無，region ='全局'，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 在Cloud DataProc群集上啟動Hive查詢作業。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **query**（_string_） - 查詢或對查詢文件的引用（q擴展名）。 * **query_uri**（_string_） - 云存儲上的hive腳本的uri。 * **variables**（_dict_） - 查詢的命名參數的映射。 * **job_name**（_string_） - DataProc集群中使用的作業名稱。默認情況下，此名稱是附加執行數據的task_id，但可以進行模板化。該名稱將始終附加一個隨機數，以避免名稱沖突。 * **cluster_name**（_string_） - DataProc集群的名稱。 * **dataproc_hive_properties**（_dict_） - Pig屬性的映射。非常適合放入默認參數 * **dataproc_hive_jars**（_list_） - 在云存儲中配置的jars的URI（例如：用于UDF和lib），非常適合放入默認參數。 * **gcp_conn_id**（_字符串_） - 用于連接到Google Cloud Platform的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **region**（_字符串_） - 創建數據加載集群的指定區域。 | | --- | --- | ##### DataProcSparkSqlOperator ``` class airflow.contrib.operators.dataproc_operator.DataProcSparkSqlOperator（query = None，query_uri = None，variables = None，job_name ='{{task.task_id}} _ {{ds_nodash}}'，cluster_name ='cluster-1'，dataproc_spark_properties =無，dataproc_spark_jars =無，gcp_conn_id ='google_cloud_default'，delegate_to =無，region ='全局'，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 在Cloud DataProc集群上啟動Spark SQL查詢作業。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **query**（_string_） - 查詢或對查詢文件的引用（q擴展名）。（模板） * **query_uri**（_string_） - 云存儲上的一個spark sql腳本的uri。 * **variables**（_dict_） - 查詢的命名參數的映射。（模板） * **job_name**（_string_） - DataProc集群中使用的作業名稱。默認情況下，此名稱是附加執行數據的task_id，但可以進行模板化。該名稱將始終附加一個隨機數，以避免名稱沖突。（模板） * **cluster_name**（_string_） - DataProc集群的名稱。（模板） * **dataproc_spark_properties**（_dict_） - Pig屬性的映射。非常適合放入默認參數 * **dataproc_spark_jars**（_list_） - 在云存儲中配置的jars的URI（例如：用于UDF和lib），非常適合放入默認參數。 * **gcp_conn_id**（_字符串_） - 用于連接到Google Cloud Platform的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **region**（_字符串_） - 創建數據加載集群的指定區域。 | | --- | --- | ##### DataProcSparkOperator ``` class airflow.contrib.operators.dataproc_operator.DataProcSparkOperator（main_jar = None，main_class = None，arguments = None，archives = None，files = None，job_name ='{{task.task_id}} _ {{ds_nodash}}'，cluster_name ='cluster-1'，dataproc_spark_properties =無，dataproc_spark_jars =無，gcp_conn_id ='google_cloud_default'，delegate_to =無，region ='全局'，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 在Cloud DataProc群集上啟動Spark作業。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **main_jar**（_字符串_） - 在云存儲上配置的作業jar的URI。（使用this或main_class，而不是兩者一起）。 * **main_class**（_string_） - 作業類的名稱。（使用this或main_jar，而不是兩者一起）。 * **arguments**（_list_） - 作業的參數。（模板） * **archives**（_list_） - 將在工作目錄中解壓縮的已歸檔文件列表。應存儲在云存儲中。 * **files**（_list_） - 要復制到工作目錄的文件列表 * **job_name**（_string_） - DataProc集群中使用的作業名稱。默認情況下，此名稱是附加執行數據的task_id，但可以進行模板化。該名稱將始終附加一個隨機數，以避免名稱沖突。（模板） * **cluster_name**（_string_） - DataProc集群的名稱。（模板） * **dataproc_spark_properties**（_dict_） - Pig屬性的映射。非常適合放入默認參數 * **dataproc_spark_jars**（_list_） - 在云存儲中配置的jars的URI（例如：用于UDF和lib），非常適合放入默認參數。 * **gcp_conn_id**（_字符串_） - 用于連接到Google Cloud Platform的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **region**（_字符串_） - 創建數據加載集群的指定區域。 | | --- | --- | ##### DataProcHadoopOperator ``` class airflow.contrib.operators.dataproc_operator.DataProcHadoopOperator（main_jar = None，main_class = None，arguments = None，archives = None，files = None，job_name ='{{task.task_id}} _ {{ds_nodash}}'，cluster_name ='cluster-1'，dataproc_hadoop_properties =無，dataproc_hadoop_jars =無，gcp_conn_id ='google_cloud_default'，delegate_to =無，region ='全局'，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 在Cloud DataProc群集上啟動Hadoop作業。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **main_jar**（_字符串_） - 在云存儲上配置的作業jar的URI。（使用this或main_class，而不是兩者一起）。 * **main_class**（_string_） - 作業類的名稱。（使用this或main_jar，而不是兩者一起）。 * **arguments**（_list_） - 作業的參數。（模板） * **archives**（_list_） - 將在工作目錄中解壓縮的已歸檔文件列表。應存儲在云存儲中。 * **files**（_list_） - 要復制到工作目錄的文件列表 * **job_name**（_string_） - DataProc集群中使用的作業名稱。默認情況下，此名稱是附加執行數據的task_id，但可以進行模板化。該名稱將始終附加一個隨機數，以避免名稱沖突。（模板） * **cluster_name**（_string_） - DataProc集群的名稱。（模板） * **dataproc_hadoop_properties**（_dict_） - Pig屬性的映射。非常適合放入默認參數 * **dataproc_hadoop_jars**（_list_） - 在云存儲中配置的jars的URI（例如：用于UDF和lib），非常適合放入默認參數。 * **gcp_conn_id**（_字符串_） - 用于連接到Google Cloud Platform的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **region**（_字符串_） - 創建數據加載集群的指定區域。 | | --- | --- | ##### DataProcPySparkOperator ``` class airflow.contrib.operators.dataproc_operator.DataProcPySparkOperator（main，arguments = None，archives = None，pyfiles = None，files = None，job_name ='{{task.task_id}} _ {{ds_nodash}}'，cluster_name =' cluster-1'，dataproc_pyspark_properties = None，dataproc_pyspark_jars = None，gcp_conn_id ='google_cloud_default'，delegate_to = None，region ='global'，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 在Cloud DataProc群集上啟動PySpark作業。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **main**（_字符串_） - [必需]用作驅動程序的主Python文件的Hadoop兼容文件系統（HCFS）URI。必須是.py文件。 * **arguments**（_list_） - 作業的參數。（模板） * **archives**（_list_） - 將在工作目錄中解壓縮的已歸檔文件列表。應存儲在云存儲中。 * **files**（_list_） - 要復制到工作目錄的文件列表 * **pyfiles**（_list_） - 要傳遞給PySpark框架的Python文件列表。支持的文件類型：.py，.egg和.zip * **job_name**（_string_） - DataProc集群中使用的作業名稱。默認情況下，此名稱是附加執行數據的task_id，但可以進行模板化。該名稱將始終附加一個隨機數，以避免名稱沖突。（模板） * **cluster_name**（_string_） - DataProc集群的名稱。 * **dataproc_pyspark_properties**（_dict_） - Pig屬性的映射。非常適合放入默認參數 * **dataproc_pyspark_jars**（_list_） - 在云存儲中配置的jars的URI（例如：用于UDF和lib），非常適合放入默認參數。 * **gcp_conn_id**（_字符串_） - 用于連接到Google Cloud Platform的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **region**（_字符串_） - 創建數據加載集群的指定區域。 | | --- | --- | ##### DataprocWorkflowTemplateInstantiateOperator ``` class airflow.contrib.operators.dataproc_operator.DataprocWorkflowTemplateInstantiateOperator（template_id，* args，** kwargs） ``` 基地： [`airflow.contrib.operators.dataproc_operator.DataprocWorkflowTemplateBaseOperator`](code.html "airflow.contrib.operators.dataproc_operator.DataprocWorkflowTemplateBaseOperator") 在Google Cloud Dataproc上實例化WorkflowTemplate。操作員將等待WorkflowTemplate完成執行。 也可以看看 請參閱：[https](https://cloud.google.com/dataproc/docs/reference/rest/v1beta2/projects.regions.workflowTemplates/instantiate)：[//cloud.google.com/dataproc/docs/reference/rest/v1beta2/projects.regions.workflowTemplates/instantiate](https://cloud.google.com/dataproc/docs/reference/rest/v1beta2/projects.regions.workflowTemplates/instantiate) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **template_id**（_string_） - 模板的id。（模板） * **project_id**（_字符串_） - 模板運行所在的Google云項目的ID * **region**（_字符串_） - 保留為“全局”，將來可能會變得相關 * **gcp_conn_id**（_字符串_） - 用于連接到Google Cloud Platform的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 | | --- | --- | ##### DataprocWorkflowTemplateInstantiateInlineOperator ``` class airflow.contrib.operators.dataproc_operator.DataprocWorkflowTemplateInstantiateInlineOperator（template，* args，** kwargs） ``` 基地： [`airflow.contrib.operators.dataproc_operator.DataprocWorkflowTemplateBaseOperator`](code.html "airflow.contrib.operators.dataproc_operator.DataprocWorkflowTemplateBaseOperator") 在Google Cloud Dataproc上實例化WorkflowTemplate內聯。操作員將等待WorkflowTemplate完成執行。 也可以看看 請參閱：[https](https://cloud.google.com/dataproc/docs/reference/rest/v1beta2/projects.regions.workflowTemplates/instantiateInline)：[//cloud.google.com/dataproc/docs/reference/rest/v1beta2/projects.regions.workflowTemplates/instantiateInline](https://cloud.google.com/dataproc/docs/reference/rest/v1beta2/projects.regions.workflowTemplates/instantiateInline) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **template**（_map_） - 模板內容。（模板） * **project_id**（_字符串_） - 模板運行所在的Google云項目的ID * **region**（_字符串_） - 保留為“全局”，將來可能會變得相關 * **gcp_conn_id**（_字符串_） - 用于連接到Google Cloud Platform的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 | | --- | --- | ### 云數據存儲區 #### 數據存儲區運營商 * [DatastoreExportOperator](28)：將實體從Google Cloud Datastore導出到云存儲。 * [DatastoreImportOperator](28)：將實體從云存儲導入Google Cloud Datastore。 ##### DatastoreExportOperator ``` class airflow.contrib.operators.datastore_export_operator.DatastoreExportOperator（bucket，namespace = None，datastore_conn_id ='google_cloud_default'，cloud_storage_conn_id ='google_cloud_default'，delegate_to = None，entity_filter = None，labels = None，polling_interval_in_seconds = 10，overwrite_existing = False，xcom_push =假，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 將實體從Google Cloud Datastore導出到云存儲 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_字符串_） - 要備份數據的云存儲桶的名稱 * **namespace**（_str_） - 指定云存儲桶中用于備份數據的可選命名空間路徑。如果GCS中不存在此命名空間，則將創建該命名空間。 * **datastore_conn_id**（_string_） - 要使用的數據存儲區連接ID的名稱 * **cloud_storage_conn_id**（_string_） - 強制寫入備份的云存儲連接ID的名稱 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **entity_filter**（_dict_） - 導出中包含項目中哪些數據的說明，請參閱[https://cloud.google.com/datastore/docs/reference/rest/Shared.Types/EntityFilter](https://cloud.google.com/datastore/docs/reference/rest/Shared.Types/EntityFilter) * **labels**（_dict_） - 客戶端分配的云存儲標簽 * **polling_interval_in_seconds**（_int_） - 再次輪詢執行狀態之前等待的秒數 * **overwrite_existing**（_bool_） - 如果存儲桶+命名空間不為空，則在導出之前將清空它。這樣可以覆蓋現有備份。 * **xcom_push**（_bool_） - 將操作名稱推送到xcom以供參考 | | --- | --- | ##### DatastoreImportOperator ``` class airflow.contrib.operators.datastore_import_operator.DatastoreImportOperator（bucket，file，namespace = None，entity_filter = None，labels = None，datastore_conn_id ='google_cloud_default'，delegate_to = None，polling_interval_in_seconds = 10，xcom_push = False，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 將實體從云存儲導入Google Cloud Datastore <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_字符串_） - 云存儲中用于存儲數據的容器 * **file**（_字符串_） - 指定云存儲桶中備份元數據文件的路徑。它應該具有擴展名.overall_export_metadata * **namespace**（_str_） - 指定云存儲桶中備份元數據文件的可選命名空間。 * **entity_filter**（_dict_） - 導出中包含項目中哪些數據的說明，請參閱[https://cloud.google.com/datastore/docs/reference/rest/Shared.Types/EntityFilter](https://cloud.google.com/datastore/docs/reference/rest/Shared.Types/EntityFilter) * **labels**（_dict_） - 客戶端分配的云存儲標簽 * **datastore_conn_id**（_string_） - 要使用的連接ID的名稱 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **polling_interval_in_seconds**（_int_） - 再次輪詢執行狀態之前等待的秒數 * **xcom_push**（_bool_） - 將操作名稱推送到xcom以供參考 | | --- | --- | #### DatastoreHook ``` class airflow.contrib.hooks.datastore_hook.DatastoreHook（datastore_conn_id ='google_cloud_datastore_default'，delegate_to = None） ``` 基地： [`airflow.contrib.hooks.gcp_api_base_hook.GoogleCloudBaseHook`](code.html "airflow.contrib.hooks.gcp_api_base_hook.GoogleCloudBaseHook") 與Google Cloud Datastore互動。此掛鉤使用Google Cloud Platform連接。 此對象不是線程安全的。如果要同時發出多個請求，則需要為每個線程創建一個鉤子。 ``` allocate_ids（partialKeys） ``` 為不完整的密鑰分配ID。請參閱[https://cloud.google.com/datastore/docs/reference/rest/v1/projects/allocateIds](https://cloud.google.com/datastore/docs/reference/rest/v1/projects/allocateIds) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **partialKeys** - 部分鍵列表 | | --- | --- | | 返回： | 完整密鑰列表。 | | --- | --- | ``` begin_transaction（） ``` 獲取新的事務處理 > 也可以看看 > > [https://cloud.google.com/datastore/docs/reference/rest/v1/projects/beginTransaction](https://cloud.google.com/datastore/docs/reference/rest/v1/projects/beginTransaction) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 返回： | 交易句柄 | | --- | --- | ``` 提交（體） ``` 提交事務，可選地創建，刪除或修改某些實體。 也可以看看 [https://cloud.google.com/datastore/docs/reference/rest/v1/projects/commit](https://cloud.google.com/datastore/docs/reference/rest/v1/projects/commit) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **body** - 提交請求的主體 | | --- | --- | | 返回： | 提交請求的響應主體 | | --- | --- | ``` delete_operation（名稱） ``` 刪除長時間運行的操作 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **name** - 操作資源的名稱 | | --- | --- | ``` export_to_storage_bucket（bucket，namespace = None，entity_filter = None，labels = None） ``` 將實體從Cloud Datastore導出到Cloud Storage進行備份 ``` get_conn（版本= 'V1'） ``` 返回Google云端存儲服務對象。 ``` GET_OPERATION（名稱） ``` 獲取長時間運行的最新狀態 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **name** - 操作資源的名稱 | | --- | --- | ``` import_from_storage_bucket（bucket，file，namespace = None，entity_filter = None，labels = None） ``` 將備份從云存儲導入云數據存儲 ``` lookup（keys，read_consistency = None，transaction = None） ``` 按鍵查找一些實體 也可以看看 [https://cloud.google.com/datastore/docs/reference/rest/v1/projects/lookup](https://cloud.google.com/datastore/docs/reference/rest/v1/projects/lookup) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **keys** - 要查找的鍵 * **read_consistency** - 要使用的讀取一致性。默認，強或最終。不能與事務一起使用。 * **transaction** - 要使用的事務，如果有的話。 | | --- | --- | | 返回： | 查找請求的響應主體。 | | --- | --- | ``` poll_operation_until_done（name，polling_interval_in_seconds） ``` 輪詢備份操作狀態直到完成 ``` 回滾（事務） ``` 回滾交易 也可以看看 [https://cloud.google.com/datastore/docs/reference/rest/v1/projects/rollback](https://cloud.google.com/datastore/docs/reference/rest/v1/projects/rollback) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **transaction** - 要回滾的事務 | | --- | --- | ``` run_query（體） ``` 運行實體查詢。 也可以看看 [https://cloud.google.com/datastore/docs/reference/rest/v1/projects/runQuery](https://cloud.google.com/datastore/docs/reference/rest/v1/projects/runQuery) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **body** - 查詢請求的主體 | | --- | --- | | 返回： | 批量查詢結果。 | | --- | --- | ### 云ML引擎 #### 云ML引擎運營商 * [MLEngineBatchPredictionOperator](28)：啟動Cloud ML Engine批量預測作業。 * [MLEngineModelOperator](28)：管理Cloud ML Engine模型。 * [MLEngineTrainingOperator](28)：啟動Cloud ML Engine培訓工作。 * [MLEngineVersionOperator](28)：管理Cloud ML Engine模型版本。 ##### MLEngineBatchPredictionOperator ``` class airflow.contrib.operators.mlengine_operator.MLEngineBatchPredictionOperator（project_id，job_id，region，data_format，input_paths，output_path，model_name = None，version_name = None，uri = None，max_worker_count = None，runtime_version = None，gcp_conn_id ='google_cloud_default'，delegate_to =無，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 啟動Google Cloud ML Engine預測作業。 注意：對于模型原點，用戶應該考慮以下三個選項中的一個：1。僅填充“uri”字段，該字段應該是指向tensorflow savedModel目錄的GCS位置。2.僅填充'model_name'字段，該字段引用現有模型，并將使用模型的默認版本。3.填充“model_name”和“version_name”字段，這些字段指特定模型的特定版本。 在選項2和3中，模型和版本名稱都應包含最小標識符。例如，打電話 ``` MLEngineBatchPredictionOperator ( ... , model_name = 'my_model' , version_name = 'my_version' , ... ) ``` 如果所需的型號版本是“projects / my_project / models / my_model / versions / my_version”。 有關參數的更多文檔，請參閱[https://cloud.google.com/ml-engine/reference/rest/v1/projects.jobs](https://cloud.google.com/ml-engine/reference/rest/v1/projects.jobs)。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **project_id**（_字符串_） - 提交預測作業的Google Cloud項目名稱。（模板） * **job_id**（_字符串_） - Google Cloud ML Engine上預測作業的唯一ID。（模板） * **data_format**（_string_） - 輸入數據的格式。如果未提供或者不是[“TEXT”，“TF_RECORD”，“TF_RECORD_GZIP”]之一，它將默認為“DATA_FORMAT_UNSPECIFIED”。 * **input_paths**（_字符串列表_） - 批量預測的輸入數據的GCS路徑列表。接受通配符運算符[*](28)，但僅限于結尾處。（模板） * **output_path**（_string_） - 寫入預測結果的GCS路徑。（模板） * **region**（_字符串_） - 用于運行預測作業的Google Compute Engine區域。（模板化） * **model_name**（_字符串_） - 用于預測的Google Cloud ML Engine模型。如果未提供version_name，則將使用此模型的默認版本。如果提供了version_name，則不應為None。如果提供uri，則應為None。（模板） * **version_name**（_字符串_） - 用于預測的Google Cloud ML Engine模型版本。如果提供uri，則應為None。（模板） * **uri**（_字符串_） - 用于預測的已保存模型的GCS路徑。如果提供了model_name，則應為None。它應該是指向張量流SavedModel的GCS路徑。（模板） * **max_worker_count**（_int_） - 用于并行處理的最大worker數。如果未指定，則默認為10。 * **runtime_version**（_字符串_） - 用于批量預測的Google Cloud ML Engine運行時版本。 * **gcp_conn_id**（_字符串_） - 用于連接到Google Cloud Platform的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用doamin范圍的委派。 | | --- | --- | ``` Raises: ``` `ValueError` ：如果無法確定唯一的模型/版本來源。 ##### MLEngineModelOperator ``` class airflow.contrib.operators.mlengine_operator.MLEngineModelOperator（project_id，model，operation ='create'，gcp_conn_id ='google_cloud_default'，delegate_to = None，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 管理Google Cloud ML Engine模型的運營商。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **project_id**（_字符串_） - MLEngine模型所屬的Google Cloud項目名稱。（模板） * **型號**（_字典_） - 包含有關模型信息的字典。如果<cite>操作</cite>是<cite>create</cite>，則<cite>model</cite>參數應包含有關此模型的所有信息，例如<cite>name</cite>。 如果<cite>操作</cite>是<cite>get</cite>，則<cite>model</cite>參數應包含<cite>模型</cite>的<cite>名稱</cite>。 * **操作** - 執行的操作。可用的操作是： * `create`：創建<cite>model</cite>參數提供的新模型。 * `get`：獲取在模型中指定名稱的特定<cite>模型</cite>。 * **gcp_conn_id**（_string_） - 獲取連接信息時使用的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 | | --- | --- | ##### MLEngineTrainingOperator ``` class airflow.contrib.operators.mlengine_operator.MLEngineTrainingOperator（project_id，job_id，package_uris，training_python_module，training_args，region，scale_tier = None，runtime_version = None，python_version = None，job_dir = None，gcp_conn_id ='google_cloud_default'，delegate_to = None，mode ='生產'，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 啟動MLEngine培訓工作的操作員。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **project_id**（_字符串_） - 應在其中運行MLEngine培訓作業的Google Cloud項目名稱（模板化）。 * **job_id**（_字符串_） - 提交的Google MLEngine培訓作業的唯一模板化ID。（模板） * **package_uris**（_字符串_） - MLEngine培訓作業的包位置列表，其中應包括主要培訓計劃+任何其他依賴項。（模板） * **training_python_module**（_string_） - 安裝'package_uris'軟件包后，在MLEngine培訓作業中運行的Python模塊名稱。（模板） * **training_args**（_string_） - 傳遞給MLEngine訓練程序的模板化命令行參數列表。（模板） * **region**（_字符串_） - 用于運行MLEngine培訓作業的Google Compute Engine區域（模板化）。 * **scale_tier**（_string_） - MLEngine培訓作業的資源層。（模板） * **runtime_version**（_字符串_） - 用于培訓的Google Cloud ML運行時版本。（模板） * **python_version**（_string_） - 訓練中使用的Python版本。（模板） * **job_dir**（_字符串_） - 用于存儲培訓輸出和培訓所需的其他數據的Google云端存儲路徑。（模板） * **gcp_conn_id**（_string_） - 獲取連接信息時使用的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **mode**（_string_） - 可以是'DRY_RUN'/'CLOUD'之一。在“DRY_RUN”模式下，不會啟動真正的培訓作業，但會打印出MLEngine培訓作業請求。在“CLOUD”模式下，將發出真正的MLEngine培訓作業創建請求。 | | --- | --- | ##### MLEngineVersionOperator ``` class airflow.contrib.operators.mlengine_operator.MLEngineVersionOperator（project_id，model_name，version_name = None，version = None，operation ='create'，gcp_conn_id ='google_cloud_default'，delegate_to = None，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 管理Google Cloud ML Engine版本的運營商。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **project_id**（_字符串_） - MLEngine模型所屬的Google Cloud項目名稱。 * **model_name**（_字符串_） - 版本所屬的Google Cloud ML Engine模型的名稱。（模板） * **version_name**（_string_） - 用于正在操作的版本的名稱。如果沒有人及<cite>版本</cite>的說法是沒有或不具備的值<cite>名稱</cite>鍵，那么這將是有效載荷中用于填充<cite>名稱</cite>鍵。（模板） * **version**（_dict_） - 包含版本信息的字典。如果<cite>操作</cite>是<cite>create</cite>，則<cite>version</cite>應包含有關此版本的所有信息，例如name和deploymentUrl。如果<cite>操作</cite>是<cite>get</cite>或<cite>delete</cite>，則<cite>version</cite>參數應包含<cite>版本</cite>的<cite>名稱</cite>。如果是None，則唯一可能的<cite>操作</cite>是<cite>list</cite>。（模板） * **操作**（_字符串_） - 執行的操作。可用的操作是： * `create`：在<cite>model_name</cite>指定的<cite>模型中</cite>創建新版本，在這種情況下，<cite>version</cite>參數應包含創建該版本的所有信息（例如<cite>name</cite>，<cite>deploymentUrl</cite>）。 * `get`：獲取<cite>model_name</cite>指定的<cite>模型中</cite>特定版本的完整信息。應在<cite>version</cite>參數中指定版本的名稱。 * `list`：列出<cite>model_name</cite>指定的<cite>模型的</cite>所有可用版本。 * `delete`：從<cite>model_name</cite>指定的<cite>模型中</cite>刪除<cite>version</cite>參數中指定的<cite>版本</cite>。應在<cite>version</cite>參數中指定版本的名稱。 * **gcp_conn_id**（_string_） - 獲取連接信息時使用的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 | | --- | --- | #### Cloud ML Engine Hook ##### MLEngineHook ``` class airflow.contrib.hooks.gcp_mlengine_hook.MLEngineHook（gcp_conn_id ='google_cloud_default'，delegate_to = None） ``` 基地： [`airflow.contrib.hooks.gcp_api_base_hook.GoogleCloudBaseHook`](code.html "airflow.contrib.hooks.gcp_api_base_hook.GoogleCloudBaseHook") ``` create_job（project_id，job，use_existing_job_fn = None） ``` 啟動MLEngine作業并等待它達到終端狀態。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **project_id**（_字符串_） - 將在其中啟動MLEngine作業的Google Cloud項目ID。 * **工作**（_字典_） - 應該提供給MLEngine API的MLEngine Job對象，例如： ``` { 'jobId' : 'my_job_id' , 'trainingInput' : { 'scaleTier' : 'STANDARD_1' , ... } } ``` * **use_existing_job_fn**（_function_） - 如果已存在具有相同job_id的MLEngine作業，則此方法（如果提供）將決定是否應使用此現有作業，繼續等待它完成并返回作業對象。它應該接受MLEngine作業對象，并返回一個布爾值，指示是否可以重用現有作業。如果未提供“use_existing_job_fn”，我們默認重用現有的MLEngine作業。 | | --- | --- | | 返回： | 如果作業成功到達終端狀態（可能是FAILED或CANCELED狀態），則為MLEngine作業對象。 | | --- | --- | | 返回類型： | 字典 | | --- | --- | ``` create_model（project_id，model） ``` 創建一個模型。阻止直到完成。 ``` create_version（project_id，model_name，version_spec） ``` 在Google Cloud ML Engine上創建版本。 如果版本創建成功則返回操作，否則引發錯誤。 ``` delete_version（project_id，model_name，version_name） ``` 刪除給定版本的模型。阻止直到完成。 ``` get_conn（） ``` 返回Google MLEngine服務對象。 ``` get_model（project_id，model_name） ``` 獲取一個模型。阻止直到完成。 ``` list_versions（project_id，model_name） ``` 列出模型的所有可用版本。阻止直到完成。 ``` set_default_version（project_id，model_name，version_name） ``` 將版本設置為默認值。阻止直到完成。 ### 云儲存 #### 存儲運營商 * [FileToGoogleCloudStorageOperator](28)：將文件上傳到Google云端存儲。 * [GoogleCloudStorageCreateBucketOperator](28)：創建新的云存儲桶。 * [GoogleCloudStorageListOperator](28)：列出存儲桶中的所有對象，并在名稱中添加字符串前綴和分隔符。 * [GoogleCloudStorageDownloadOperator](28)：從Google云端存儲中下載文件。 * [GoogleCloudStorageToBigQueryOperator](28)：將Google云存儲中的文件加載到BigQuery中。 * [GoogleCloudStorageToGoogleCloudStorageOperator](28)：將對象從存儲桶復制到另一個存儲桶，并在需要時重命名。 ##### FileToGoogleCloudStorageOperator ``` class airflow.contrib.operators.file_to_gcs.FileToGoogleCloudStorageOperator（src，dst，bucket，google_cloud_storage_conn_id ='google_cloud_default'，mime_type ='application / octet-stream'，delegate_to = None，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 將文件上傳到Google云端存儲 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **src**（_string_） - 本地文件的路徑。（模板） * **dst**（_string_） - 指定存儲桶中的目標路徑。（模板） * **bucket**（_string_） - 要上傳的存儲桶。（模板） * **google_cloud_storage_conn_id**（_字符串_） - 要上傳的Airflow連接ID * **mime_type**（_string_） - mime類型字符串 * **delegate_to**（_string_） - 模擬的帳戶（如果有） | | --- | --- | ``` 執行（上下文） ``` 將文件上傳到Google云端存儲 ##### GoogleCloudStorageCreateBucketOperator ``` class airflow.contrib.operators.gcs_operator.GoogleCloudStorageCreateBucketOperator（bucket_name，storage_class ='MULTI_REGIONAL'，location ='US'，project_id = None，labels = None，google_cloud_storage_conn_id ='google_cloud_default'，delegate_to = None，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 創建一個新存儲桶。Google云端存儲使用平面命名空間，因此您無法創建名稱已在使用中的存儲桶。 > 也可以看看 > > 有關詳細信息，請參閱存儲桶命名指南：[https](https://cloud.google.com/storage/docs/bucketnaming.html)：[//cloud.google.com/storage/docs/bucketnaming.html#requirements](https://cloud.google.com/storage/docs/bucketnaming.html) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket_name**（_string_） - 存儲桶的名稱。（模板） * **storage_class**（_string_） - 這定義了存儲桶中對象的存儲方式，并確定了SLA和存儲成本（模板化）。價值包括 * `MULTI_REGIONAL` * `REGIONAL` * `STANDARD` * `NEARLINE` * `COLDLINE` 。 如果在創建存儲桶時未指定此值，則默認為STANDARD。 * **位置**（_字符串_） - 水桶的位置。（模板化）存儲桶中對象的對象數據駐留在此區域內的物理存儲中。默認為美國。 也可以看看 [https://developers.google.com/storage/docs/bucket-locations](https://developers.google.com/storage/docs/bucket-locations) * **project_id**（_string_） - GCP項目的ID。（模板） * **labels**（_dict_） - 用戶提供的鍵/值對標簽。 * **google_cloud_storage_conn_id**（_字符串_） - 連接到Google云端存儲時使用的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 | | --- | --- | ``` Example: ``` 以下運算符將在區域中創建`test-bucket`具有`MULTI_REGIONAL`存儲類的新存儲桶`EU` ``` CreateBucket = GoogleCloudStorageCreateBucketOperator ( task_id = 'CreateNewBucket' , bucket_name = 'test-bucket' , storage_class = 'MULTI_REGIONAL' , location = 'EU' , labels = { 'env' : 'dev' , 'team' : 'airflow' }, google_cloud_storage_conn_id = 'airflow-service-account' ) ``` ##### GoogleCloudStorageDownloadOperator ``` class airflow.contrib.operators.gcs_download_operator.GoogleCloudStorageDownloadOperator（bucket，object，filename = None，store_to_xcom_key = None，google_cloud_storage_conn_id ='google_cloud_default'，delegate_to = None，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 從Google云端存儲下載文件。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_字符串_） - 對象所在的Google云存儲桶。（模板） * **object**（_字符串_） - 要在Google云存儲桶中下載的對象的名稱。（模板） * **filename**（_字符串_） - 應將文件下載到的本地文件系統（正在執行操作符的位置）上的文件路徑。（模板化）如果未傳遞文件名，則下載的數據將不會存儲在本地文件系統中。 * **store_to_xcom_key**（_string_） - 如果設置了此參數，操作員將使用此參數中設置的鍵將下載文件的內容推送到XCom。如果未設置，則下載的數據不會被推送到XCom。（模板） * **google_cloud_storage_conn_id**（_字符串_） - 連接到Google云端存儲時使用的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 | | --- | --- | ##### GoogleCloudStorageListOperator ``` class airflow.contrib.operators.gcs_list_operator.GoogleCloudStorageListOperator（bucket，prefix = None，delimiter = None，google_cloud_storage_conn_id ='google_cloud_default'，delegate_to = None，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 使用名稱中的給定字符串前綴和分隔符列出存儲桶中的所有對象。 ``` 此運算符返回一個python列表，其中包含可供其使用的對象的名稱 ``` <cite>xcom</cite>在下游任務中。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_字符串_） - 用于查找對象的Google云存儲桶。（模板） * **prefix**（_string_） - 前綴字符串，用于過濾名稱以此前綴開頭的對象。（模板） * **delimiter**（_string_） - 要過濾對象的分隔符。（模板化）例如，要列出GCS目錄中的CSV文件，您可以使用delimiter ='。csv'。 * **google_cloud_storage_conn_id**（_字符串_） - 連接到Google云端存儲時使用的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 | | --- | --- | ``` Example: ``` 以下運算符將列出存儲桶中文件`sales/sales-2017`夾中的所有Avro文件`data`。 ``` GCS_Files = GoogleCloudStorageListOperator ( task_id = 'GCS_Files' , bucket = 'data' , prefix = 'sales/sales-2017/' , delimiter = '.avro' , google_cloud_storage_conn_id = google_cloud_conn_id ) ``` ##### GoogleCloudStorageToBigQueryOperator ``` class airflow.contrib.operators.gcs_to_bq.GoogleCloudStorageToBigQueryOperator（bucket，source_objects，destination_project_dataset_table，schema_fields = None，schema_object = None，source_format ='CSV'，compression ='NONE'，create_disposition ='CREATE_IF_NEEDED'，skip_leading_rows = 0，write_disposition =' WRITE_EMPTY'，field_delimiter ='，'，max_bad_records = 0，quote_character = None，ignore_unknown_values = False，allow_quoted_newlines = False，allow_jagged_rows = False，max_id_key = None，bigquery_conn_id ='bigquery_default'，google_cloud_storage_conn_id ='google_cloud_default'，delegate_to = None，schema_update_options =（），src_fmt_configs = {}，external_table = False，time_partitioning = {}，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 將文件從Google云存儲加載到BigQuery中。 可以用兩種方法之一指定用于BigQuery表的模式。您可以直接傳遞架構字段，也可以將運營商指向Google云存儲對象名稱。Google云存儲中的對象必須是包含架構字段的JSON文件。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_string_） - 要加載的桶。（模板） * **source_objects** - 要加載的Google云存儲URI列表。（模板化）如果source_format是'DATASTORE_BACKUP'，則列表必須只包含一個URI。 * **destination_project_dataset_table**（_string_） - 用于加載數據的虛線（<project>。）<dataset>。<table> BigQuery表。如果未包含<project>，則項目將是連接json中定義的項目。（模板） * **schema_fields**（_list_） - 如果設置，則此處定義的架構字段列表：[https](https://cloud.google.com/bigquery/docs/reference/v2/jobs)：**//cloud.google.com/bigquery/docs/reference/v2/jobs#configuration.load**當source_format為'DATASTORE_BACKUP'時，不應設置。 * **schema_object** - 如果設置，則指向包含表的架構的.json文件的GCS對象路徑。（模板） * **schema_object** - 字符串 * **source_format**（_string_） - 要導出的文件格式。 * **compression**（_字符串_） - [可選]數據源的壓縮類型。可能的值包括GZIP和NONE。默認值為NONE。Google Cloud Bigtable，Google Cloud Datastore備份和Avro格式會忽略此設置。 * **create_disposition**（_string_） - 如果表不存在，則創建處置。 * **skip_leading_rows**（_int_） - 從CSV加載時要跳過的行數。 * **write_disposition**（_string_） - 表已存在時的寫處置。 * **field_delimiter**（_string_） - 從CSV加載時使用的分隔符。 * **max_bad_records**（_int_） - BigQuery在運行作業時可以忽略的最大錯誤記錄數。 * **quote_character**（_string_） - 用于引用CSV文件中數據部分的值。 * **ignore_unknown_values**（_bool_） - [可選]指示BigQuery是否應允許表模式中未表示的額外值。如果為true，則忽略額外值。如果為false，則將具有額外列的記錄視為錯誤記錄，如果錯誤記錄太多，則在作業結果中返回無效錯誤。 * **allow_quoted_newlines**（_boolean_） - 是否允許引用的換行符（true）或不允許（false）。 * **allow_jagged_rows**（_bool_） - 接受缺少尾隨可選列的行。缺失值被視為空值。如果為false，則缺少尾隨列的記錄將被視為錯誤記錄，如果錯誤記錄太多，則會在作業結果中返回無效錯誤。僅適用于CSV，忽略其他格式。 * **max_id_key**（_string_） - 如果設置，則是BigQuery表中要加載的列的名稱。在加載發生后，Thsi將用于從BigQuery中選擇MAX值。結果將由execute（）命令返回，該命令又存儲在XCom中供將來的操作員使用。這對增量加載很有幫助 - 在將來的執行過程中，您可以從最大ID中獲取。 * **bigquery_conn_id**（_string_） - 對特定BigQuery掛鉤的引用。 * **google_cloud_storage_conn_id**（_字符串_） - 對特定Google云存儲掛鉤的引用。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 * **schema_update_options**（_list_） - 允許更新目標表的模式作為加載作業的副作用。 * **src_fmt_configs**（_dict_） - 配置特定于源格式的可選字段 * **external_table**（_bool_） - 用于指定目標表是否應為BigQuery外部表的標志。默認值為False。 * **time_partitioning**（_dict_） - 配置可選的時間分區字段，即按API規范按字段，類型和到期分區。請注意，“field”在dataset.table $ partition的并發中不可用。 | | --- | --- | ##### GoogleCloudStorageToGoogleCloudStorageOperator ``` class airflow.contrib.operators.gcs_to_gcs.GoogleCloudStorageToGoogleCloudStorageOperator（source_bucket，source_object，destination_bucket = None，destination_object = None，move_object = False，google_cloud_storage_conn_id ='google_cloud_default'，delegate_to = None，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") 將對象從存儲桶復制到另一個存儲桶，并在需要時重命名。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **source_bucket**（_string_） - 對象所在的源Google云存儲桶。（模板） * **source_object**（_string_） - 要在Google云存儲分區中復制的對象的源名稱。（模板化）如果在此參數中使用通配符： > 您只能在存儲桶中使用一個通配符作為對象（文件名）。通配符可以出現在對象名稱內或對象名稱的末尾。不支持在存儲桶名稱中附加通配符。 * **destination_bucket** - 目標Google云端存儲分區 | | --- | --- | 對象應該在哪里。（模板化）：type destination_bucket：string：param destination_object：對象的目標名稱 > 目標Google云存儲桶。（模板化）如果在source_object參數中提供了通配符，則這是將添加到最終目標對象路徑的前綴。請注意，將刪除通配符之前的源路徑部分; 如果需要保留，則應將其附加到destination_object。例如，使用prefix `foo/*`和destination_object'blah <cite>/`</cite>，文件`foo/baz`將被復制到`blah/baz`; 保留前綴寫入destination_object，例如`blah/foo`，在這種情況下，復制的文件將被命名`blah/foo/baz`。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **move_object** - 當移動對象為True時，移動對象 | | --- | --- | ``` 復制到新位置。 ``` 這相當于mv命令而不是cp命令。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **google_cloud_storage_conn_id**（_字符串_） - 連接到Google云端存儲時使用的連接ID。 * **delegate_to**（_string_） - 模擬的帳戶（如果有）。為此，發出請求的服務帳戶必須啟用域范圍委派。 | | --- | --- | ``` Examples: ``` 下面的操作將命名一個文件復制`sales/sales-2017/january.avro`在`data`桶的文件和名為斗`copied_sales/2017/january-backup.avro` in the ``data_backup` ``` copy_single_file = GoogleCloudStorageToGoogleCloudStorageOperator ( task_id = 'copy_single_file' , source_bucket = 'data' , source_object = 'sales/sales-2017/january.avro' , destination_bucket = 'data_backup' , destination_object = 'copied_sales/2017/january-backup.avro' , google_cloud_storage_conn_id = google_cloud_conn_id ) ``` 以下運算符會將文件`sales/sales-2017`夾中的所有Avro文件（即名稱以該前綴開頭）復制到存儲`data`桶中的`copied_sales/2017`文件夾中`data_backup`。 ``` copy_files = GoogleCloudStorageToGoogleCloudStorageOperator ( task_id = 'copy_files' , source_bucket = 'data' , source_object = 'sales/sales-2017/*.avro' , destination_bucket = 'data_backup' , destination_object = 'copied_sales/2017/' , google_cloud_storage_conn_id = google_cloud_conn_id ) ``` 以下運算符會將文件`sales/sales-2017`夾中的所有Avro文件（即名稱以該前綴開頭）移動到`data`存儲桶中的同一文件夾`data_backup`，刪除過程中的原始文件。 ``` move_files = GoogleCloudStorageToGoogleCloudStorageOperator ( task_id = 'move_files' , source_bucket = 'data' , source_object = 'sales/sales-2017/*.avro' , destination_bucket = 'data_backup' , move_object = True , google_cloud_storage_conn_id = google_cloud_conn_id ) ``` #### GoogleCloudStorageHook ``` class airflow.contrib.hooks.gcs_hook.GoogleCloudStorageHook（google_cloud_storage_conn_id ='google_cloud_default'，delegate_to = None） ``` 基地： [`airflow.contrib.hooks.gcp_api_base_hook.GoogleCloudBaseHook`](code.html "airflow.contrib.hooks.gcp_api_base_hook.GoogleCloudBaseHook") 與Google云端存儲互動。此掛鉤使用Google Cloud Platform連接。 ``` copy（source_bucket，source_object，destination_bucket = None，destination_object = None） ``` 將對象從存儲桶復制到另一個存儲桶，并在需要時重命名。 destination_bucket或destination_object可以省略，在這種情況下使用源桶/對象，但不能同時使用兩者。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **source_bucket**（_string_） - 要從中復制的對象的存儲桶。 * **source_object**（_string_） - 要復制的對象。 * **destination_bucket**（_string_） - 要復制到的對象的目標。可以省略; 然后使用相同的桶。 * **destination_object** - 給定對象的（重命名）路徑。可以省略; 然后使用相同的名稱。 | | --- | --- | ``` create_bucket（bucket_name，storage_class ='MULTI_REGIONAL'，location ='US'，project_id = None，labels = None） ``` 創建一個新存儲桶。Google云端存儲使用平面命名空間，因此您無法創建名稱已在使用中的存儲桶。 也可以看看 有關詳細信息，請參閱存儲桶命名指南：[https](https://cloud.google.com/storage/docs/bucketnaming.html)：[//cloud.google.com/storage/docs/bucketnaming.html#requirements](https://cloud.google.com/storage/docs/bucketnaming.html) <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket_name**（_string_） - 存儲桶的名稱。 * **storage_class**（_string_） - 這定義了存儲桶中對象的存儲方式，并確定了SLA和存儲成本。價值包括 * `MULTI_REGIONAL` * `REGIONAL` * `STANDARD` * `NEARLINE` * `COLDLINE` 。 如果在創建存儲桶時未指定此值，則默認為STANDARD。 * **位置**（_字符串_） - 水桶的位置。存儲桶中對象的對象數據駐留在此區域內的物理存儲中。默認為美國。 也可以看看 [https://developers.google.com/storage/docs/bucket-locations](https://developers.google.com/storage/docs/bucket-locations) * **project_id**（_string_） - GCP項目的ID。 * **labels**（_dict_） - 用戶提供的鍵/值對標簽。 | | --- | --- | | 返回： | 如果成功，則返回`id`桶的內容。 | | --- | --- | ``` 刪除（桶，對象，生成=無） ``` 如果未對存儲桶啟用版本控制，或者使用了生成參數，則刪除對象。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_字符串_） - 對象所在的存儲桶的名稱 * **object**（_字符串_） - 要刪除的對象的名稱 * **generation**（_字符串_） - 如果存在，則永久刪除該代的對象 | | --- | --- | | 返回： | 如果成功則為真 | | --- | --- | ``` 下載（bucket，object，filename = None） ``` 從Google云端存儲中獲取文件。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_string_） - 要獲取的存儲桶。 * **object**（_string_） - 要獲取的對象。 * **filename**（_string_） - 如果設置，則應寫入文件的本地文件路徑。 | | --- | --- | ``` 存在（桶，對象） ``` 檢查Google云端存儲中是否存在文件。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_字符串_） - 對象所在的Google云存儲桶。 * **object**（_字符串_） - 要在Google云存儲分區中檢查的對象的名稱。 | | --- | --- | ``` get_conn（） ``` 返回Google云端存儲服務對象。 ``` get_crc32c（bucket，object） ``` 獲取Google Cloud Storage中對象的CRC32c校驗和。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_字符串_） - 對象所在的Google云存儲桶。 * **object**（_字符串_） - 要在Google云存儲分區中檢查的對象的名稱。 | | --- | --- | ``` get_md5hash（bucket，object） ``` 獲取Google云端存儲中對象的MD5哈希值。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_字符串_） - 對象所在的Google云存儲桶。 * **object**（_字符串_） - 要在Google云存儲分區中檢查的對象的名稱。 | | --- | --- | ``` get_size（bucket，object） ``` 獲取Google云端存儲中文件的大小。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_字符串_） - 對象所在的Google云存儲桶。 * **object**（_字符串_） - 要在Google云存儲分區中檢查的對象的名稱。 | | --- | --- | ``` is_updated_after（bucket，object，ts） ``` 檢查Google Cloud Storage中是否更新了對象。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_字符串_） - 對象所在的Google云存儲桶。 * **object**（_字符串_） - 要在Google云存儲分區中檢查的對象的名稱。 * **ts**（_datetime_） - 要檢查的時間戳。 | | --- | --- | ``` list（bucket，versions = None，maxResults = None，prefix = None，delimiter = None） ``` 使用名稱中的給定字符串前綴列出存儲桶中的所有對象 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_字符串_） - 存儲桶名稱 * **versions**（_boolean_） - 如果為true，則列出對象的所有版本 * **maxResults**（_整數_） - 在單個響應頁面中返回的最大項目數 * **prefix**（_string_） - 前綴字符串，用于過濾名稱以此前綴開頭的對象 * **delimiter**（_string_） - 根據分隔符過濾對象（例如'.csv'） | | --- | --- | | 返回： | 與過濾條件匹配的對象名稱流 | | --- | --- | ``` 重寫（source_bucket，source_object，destination_bucket，destination_object = None） ``` 具有與復制相同的功能，除了可以處理超過5 TB的文件，以及在位置和/或存儲類之間復制時。 destination_object可以省略，在這種情況下使用source_object。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **source_bucket**（_string_） - 要從中復制的對象的存儲桶。 * **source_object**（_string_） - 要復制的對象。 * **destination_bucket**（_string_） - 要復制到的對象的目標。 * **destination_object** - 給定對象的（重命名）路徑。可以省略; 然后使用相同的名稱。 | | --- | --- | ``` upload（bucket，object，filename，mime_type ='application / octet-stream'） ``` 將本地文件上傳到Google云端存儲。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **bucket**（_string_） - 要上傳的存儲桶。 * **object**（_字符串_） - 上載本地文件時要設置的對象名稱。 * **filename**（_字符串_） - 要上載的文件的本地文件路徑。 * **mime_type**（_string_） - 上載文件時要設置的MIME類型。 | | --- | --- | ### 谷歌Kubernetes引擎 #### Google Kubernetes引擎集群運營商 * [GKEClusterDeleteOperator](28)：在Google Cloud Platform中創建Kubernetes群集 * [GKEPodOperator](28)：刪除Google Cloud Platform中的Kubernetes群集 ##### GKEClusterCreateOperator ``` class airflow.contrib.operators.gcp_container_operator.GKEClusterCreateOperator（project_id，location，body = {}，gcp_conn_id ='google_cloud_default'，api_version ='v2'，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") ##### GKEClusterDeleteOperator ``` class airflow.contrib.operators.gcp_container_operator.GKEClusterDeleteOperator（project_id，name，location，gcp_conn_id ='google_cloud_default'，api_version ='v2'，* args，** kwargs） ``` 基地： [`airflow.models.BaseOperator`](code.html "airflow.models.BaseOperator") ##### GKEPodOperator #### Google Kubernetes Engine Hook ``` class airflow.contrib.hooks.gcp_container_hook.GKEClusterHook（project_id，location） ``` 基地： `airflow.hooks.base_hook.BaseHook` ``` create_cluster（cluster，retry = <object object>，timeout = <object object>） ``` 創建一個群集，由指定數量和類型的Google Compute Engine實例組成。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **cluster**（_dict __或_ _google.cloud.container_v1.types.Cluster_） - 群集protobuf或dict。如果提供了dict，它必須與protobuf消息的格式相同google.cloud.container_v1.types.Cluster * **重試**（_google.api_core.retry.Retry_） - 用于重試請求的重試對象（google.api_core.retry.Retry）。如果指定None，則不會重試請求。 * **timeout**（_float_） - 等待請求完成的時間（以秒為單位）。請注意，如果指定了重試，則超時適用于每次單獨嘗試。 | | --- | --- | | 返回： | 新集群或現有集群的完整URL | | --- | --- | ``` ：加薪 ``` ParseError：在嘗試轉換dict時出現JSON解析問題AirflowException：cluster不是dict類型也不是Cluster proto類型 ``` delete_cluster（name，retry = <object object>，timeout = <object object>） ``` 刪除集群，包括Kubernetes端點和所有工作節點。在群集創建期間配置的防火墻和路由也將被刪除。群集可能正在使用的其他Google Compute Engine資源（例如，負載均衡器資源）如果在初始創建時不存在，則不會被刪除。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | * **name**（_str_） - 要刪除的集群的名稱 * **重試**（_google.api_core.retry.Retry_） - 重_試用_于確定何時/是否重試請求的對象。如果指定None，則不會重試請求。 * **timeout**（_float_） - 等待請求完成的時間（以秒為單位）。請注意，如果指定了重試，則超時適用于每次單獨嘗試。 | | --- | --- | | 返回： | 如果成功則刪除操作的完整URL，否則為None | | --- | --- | ``` get_cluster（name，retry = <object object>，timeout = <object object>） ``` 獲取指定集群的詳細信息：param name：要檢索的集群的名稱：type name：str：param retry：用于重試請求的重試對象。如果指定了None， > 請求不會被重試。 <colgroup><col class="field-name"><col class="field-body"></colgroup> | 參數： | **timeout**（_float_） - 等待請求完成的時間（以秒為單位）。請注意，如果指定了重試，則超時適用于每次單獨嘗試。 | | --- | --- | | 返回： | 一個google.cloud.container_v1.types.Cluster實例 | | --- | --- | ``` GET_OPERATION（OPERATION_NAME） ``` 從Google Cloud獲取操作：param operation_name：要獲取的操作的名稱：type operation_name：str：return：來自Google Cloud的新的更新操作 ``` wait_for_operation（操作） ``` 給定操作，持續從Google Cloud獲取狀態，直到完成或發生錯誤：param操作：等待的操作：鍵入操作：google.cloud.container_V1.gapic.enums.Operator：return：a new，updated從Google Cloud獲取的操作 ![Google 翻譯](https://www.gstatic.com/images/branding/product/1x/translate_24dp.png) # 原文提供更好的翻譯建議 * * *