Import csv files from GCS and transform using Dataflow then sink to BigQuery using Airflow sensors

Nhu Dao 0
Viewed 20

I am a newbie in Cloud Composer. My use case is: There are several files type in GCS like json, csv, txt,.. but I only want to select csv file, use Dataflow in Python to transform them (such as rename fields,...), then write it to BigQuery. And the main requirement is use Airflow sensors to trigger them whenever a new csv file import to GCS.

Do anyone know how to write it using Cloud Composer - Airflow sensors (no cloud function here): GCS (only csv file) - Dataflow - BigQuery

google-cloud-dataflow google-cloud-composer google-cloud-storage airflow google-bigquery
1 Answers

For Airflow sensor, you can use google cloud airflow sensor for GCS to detect if there is a new file in the GCS folder.

For the processing part, you can use ReadFromText and make sure to pick only CSV files. If you need to process only the new CSV files, then you can past the list of the new files detected by the sensor to the dataflow job as parameter.

For the dataflow job from composer, I suggest to build your job in a flex template, and then you can trigger the flex template from composer using DataflowStartFlexTemplateOperator. Here is an example on how to use it:

my_dataflow_job = DataflowStartFlexTemplateOperator(
        location="europe-west1",
        task_id="my-task-id",
        body={
            "launchParameter": {
                "containerSpecGcsPath": "gcs_path_to_the_flex_template_json_path",
                "jobName": "my-job-name",
                "environment": {
                    "serviceAccountEmail": "xxx",
                },
                "parameters": {
                    "key": "val",
                    "key" : "val"
                },
            }
        },
    )

And for the saving part to BigQuery, it should take part of your dataflow job

Idhem0
Related