0

I need help/suggestions to implement the below use case.

I have 2 csv files in Google Cloud Storage bucket, I need to join these 2 files, based on one common column, and I need to save the output file back into the Google Cloud Storage bucket.

I need to implement this using any Google Cloud solution (cloud data flow with beam python), Cloud function or any other Cloud solutions, since I am new to Google cloud platform, I request all to help me on implementing this use case.

Looking forward to hearing from you

Thanks in advance

1 Answer 1

1

You have several way to achieve this. If the result of the merge take less than 1Gb and you want only 1 output file, you can do like this

  • Query external CSV files from BigQuery (federated query) and save the result in a temporary table like this
CREATE OR REPLACE EXTERNAL TABLE mydataset.table1
OPTIONS (
  format = 'CSV',
  uris = ['gs://mybucket/file1.csv'],
  skip_leading_rows = 1
)

CREATE OR REPLACE EXTERNAL TABLE mydataset.table2
OPTIONS (
  format = 'CSV',
  uris = ['gs://mybucket/file2.csv'],
  skip_leading_rows = 1
)

CREATE TABLE mydataset.newtable
OPTIONS(
  expiration_timestamp=TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)
) AS
SELECT * 
FROM mydataset.table1 join mydataset.table2 ON ....

Else, you can use the solution that I describe in this article (that I wrote)


EDIT 1

You can use this sample of workflow definition that do what you need.


- loadFile1:
    call: http.post
    args:
      url: https://bigquery.googleapis.com/bigquery/v2/projects/<projectID>/jobs
      auth:
        type: OAuth2
      body:
        configuration:
          query:
            query: CREATE OR REPLACE EXTERNAL TABLE mydataset.table1 OPTIONS (format = 'CSV', uris = ['gs://mybucket/file1.csv'], skip_leading_rows = 1)
            useLegacySql: false
- loadFile2:
    call: http.post
    args:
      url: https://bigquery.googleapis.com/bigquery/v2/projects/<projectID>/jobs
      auth:
        type: OAuth2
      body:
        configuration:
          query:
            query: CREATE OR REPLACE EXTERNAL TABLE mydataset.table2 OPTIONS (format = 'CSV', uris = ['gs://mybucket/file2.csv'], skip_leading_rows = 1)
            useLegacySql: false
- joinQuery:
    call: http.post
    args:
      url: https://bigquery.googleapis.com/bigquery/v2/projects/<projectID>/jobs
      auth:
        type: OAuth2
      body:
        configuration:
          query:
            query: CREATE TABLE mydataset.newtable OPTIONS( expiration_timestamp=TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)) AS SELECT * ......
            useLegacySql: false
    result: queryResult
- getState:
    call: http.get
    args:
      url: ${"https://bigquery.googleapis.com/bigquery/v2/projects/<projectID>/jobs/" + queryResult.body.jobReference.jobId}
      auth:
        type: OAuth2
    result: jobState
    next: testState
- testState:
    switch:
      - condition: ${jobState.body.status.state == "DONE"}
        next: extractData
    next: waitAndGetState
- waitAndGetState:
    call: sys.sleep
    args:
      seconds: 1
    next: getState
- extractData:
    call: http.post
    args:
      url: https://bigquery.googleapis.com/bigquery/v2/projects/<projectID>/jobs
      auth:
        type: OAuth2
      body:
        configuration:
          extract:
            destinationUri: gs://<YourBucket>/bq-extract.csv
            destinationFormat: CSV
            sourceTable:
              projectId: <projectID>
              datasetId: mydataset
              tableId: newtable
    result: extractResult
- returnOutput:
    return: ${extractResult}

Then, use Cloud Scheduler to call directly the Create Workflow execution API with an empty body {} and a OAuth2 authentication mode.

7
  • Thanks for your answer, I can’t use Bigquery, it would be helpful if you provide solution with apache beam python (cloud data flow as a runner). The preferred cloud services are cloud storage , Apache beam with data flow as runner
    – Sahana
    Mar 27, 2021 at 19:45
  • Why you can't use BigQuery? Google itself say that, when you can use BigQuery, it's always faster and cheaper than any other solution (Beam, Hadoop or other bigdata processor) Mar 27, 2021 at 20:21
  • Thanks for your response, i can use Bigquery export data, but i need to run this on daily basis then how can i schedule Biquery export data on Daily, Do we have any option to schedule/Automate your solution...Looking forward to hear for help..Thanks in advance !!
    – Sahana
    Mar 29, 2021 at 18:08
  • Yes, you can run scheduled queries with BigQuery. But, because you have several request to run, I recommend you to use Google Workflows to orchestrate the different call. Then, you can use Cloud Scheduler to run a Workflow execution on daily basis. Let me know if you need more guidance on that option Mar 29, 2021 at 18:29
  • Thanks for your quick response. I have file1.csv & file2.csv files in gcs bucket , then i need to join these 2 files based on one common column,and then i need to write output file back into cloud storage bucket. I have to run above answered 3 queries on daily basis, please help me how can i schedule/automate these Biquery exports on Google Workflows with Cloud Scheduler.
    – Sahana
    Mar 29, 2021 at 20:03

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.