在 Airflow 上使用通配符 GoogleCloudStorageToBigQueryOperator答案

【问题标题】：Using Wild Card on Airflow GoogleCloudStorageToBigQueryOperator在 Airflow 上使用通配符 GoogleCloudStorageToBigQueryOperator
【发布时间】：2022-11-30 19:23:57
【问题描述】：

是否可以在 GoogleCloudStorageToBigQueryOperator 上使用通配符？

所以我在 GCS 的某个文件夹中有一组文件

file_sample_1.json
file_sample_2.json
file_sample_3.json
...
file_sample_n.json

我想使用带有 GoogleCloudStorageToBigQueryOperator 的气流来摄取这些文件。

下面是我的代码：

    def create_operator_write_init():
        return GoogleCloudStorageToBigQueryOperator(
            task_id = 'test_ingest_to_bq',
            bucket = 'sample-bucket-dev-202211',
            source_objects = 'file_sample_1.json',
            destination_project_dataset_table = 'sample_destination_table',
            create_disposition = "CREATE_IF_NEEDED",
            write_disposition = "WRITE_TRUNCATE",
            source_format = "NEWLINE_DELIMITED_JSON",
            schema_fields = [
                {"name": "id", "type": "INTEGER", "mode": "NULLABLE"},
                {"name": "created_at", "type": "TIMESTAMP", "mode": "NULLABLE"},
                {"name": "updated_at", "type": "TIMESTAMP", "mode": "NULLABLE"},
            ]
        )

它可以很好地摄取 1 个文件，但我需要 source_object 有通配符，我可以做类似“file_sample_*.json”的操作，以便 * 充当通配符吗？

【问题讨论】：

标签： python airflow

【解决方案1】：

是的，但您应该将字符串包含在列表中。所以如果你使用

source_objects = ['file_sample_*.json'],

它将摄取所有以“file_sample_”开头并以“.json”结尾的文件。

【讨论】：