Airflow

Question

documentation 指出 source_objects 参数采用模板化值。但是，当我尝试以下操作时：

gcs_to_bq_op = GoogleCloudStorageToBigQueryOperator(
    task_id=name,
    bucket='gdbm-public',
    source_objects=['entity/{{ ds_nodash }}.0.{}.json'.format(filename)],
    destination_project_dataset_table='dbm_public_entity.{}'.format(name),
    schema_fields=schema,
    source_format='NEWLINE_DELIMITED_JSON',
    create_disposition='CREATE_IF_NEEDED',
    write_disposition='WRITE_TRUNCATE',
    max_bad_records=0,
    allow_jagged_rows=True,
    google_cloud_storage_conn_id='my_gcp_conn',
    bigquery_conn_id='my_gcp_conn',
    delegate_to=SERVICE_ACCOUNT,
    dag=dag
    )

我收到错误信息： Exception: BigQuery job failed. Final error was: {u'reason': u'notFound', u'message': u'Not found: URI gs://gdbm-public/entity/{ ds_nodash }.0.GeoLocation.json'}.

我找到了一个 example，其中 {{ ds_nodash }} 变量以相同的方式使用。所以我不确定为什么这对我不起作用。

Answer 1

问题是在字符串上调用 .format 会导致一组双括号被删除：

>>> 'entity/{{ ds_nodash }}.0.{}.json'.format(filename)
'entity/{ ds_nodash }.0.foobar.json'

您需要通过加倍来转义您希望在字符串中的大括号：

>>> 'entity/{{{{ ds_nodash }}}}.0.{}.json'.format(filename)
'entity/{{ ds_nodash }}.0.foobar.json'

Answer 2

问题与 Dustin 所描述的完全一样，在字符串上调用 .format 会导致一组双括号被删除。但是，不是将括号加倍，而是 1 个解决方案：

'entity/{{{{ ds_nodash }}}}.0.{}.json'.format(filename)

我发现用这种方式格式化字符串更容易避免混淆：

"entity/{0}.0.{1}.json".format("{{ ds_nodash }}", filename)

Airflow - GoogleCloudStorageToBigQueryOperator 不呈现模板化 source_objects

Airflow - GoogleCloudStorageToBigQueryOperator does not render templated source_objects

python

google-cloud-platform

google-cloud-composer