У меня есть задание облачного потока данных, чтобы вытащить данные из Oracle в Google Cloud SQL в проекте GCP с настройкой VPC / VPN. Запланировал это задание потока данных, используя Composer, и выполнял его периодически. Это работало нормально, однако иногда Dataflow Job выполняется успешно, но данные не перемещаются в Cloud SQL. Ничего не ошибка / указать c события из журналов. Это странно ... Это случается очень редко ... Было бы хорошо, если бы кто-нибудь дал предложение выяснить причину root этого случая.
Вот определение оператора потока данных,
dataFlowStatus = DataFlowJavaOperator( task_id='dataflowlabelstatus', jar=jar_pathr, gcp_conn_id='google_cloud_default', options={ 'project': project_id, 'region': gcp_region, 'usePublicIps': ip_condition, 'stagingLocation': staging_location, 'tempLocation': staging_location, 'subnetwork': gcp_sub_network, 'RTMaxDate': "{{ task_instance.xcom_pull(task_ids='validateDate', key='job_last_fetch_date') }}", 'oracleJDBCSource': oracle_connection_url, 'oracleUserName': rm_username, 'oraclePassword': rm_password, 'postgreUserName_RT': receipt_username, 'postgrePassword_RT': receipt_password, 'postgresDatabaseName_RT': receipt_schema, 'postgreUserName_Label': label_username, 'postgrePassword_Label': label_password, 'postgresDatabaseName_Label': label_schema, 'cloudSqlInstance_RT': receipt_cloud_instance, 'cloudSqlInstance_Label': label_cloud_instance, 'autoscalingAlgorithm': 'BASIC', 'maxNumWorkers': '50' }, dag=dag )