Adding sample python DAG and gcs_to_gcs operator as plugin #1678

dsdinter · 2018-08-31T13:24:27Z

Author DAG, including master csv file sample and test:

Uses gcs_to_gcs operator from plugins
Uses Cloud logging features
Dynamically generates tasks based on master csv file

googlebot · 2018-08-31T13:24:30Z

We found a Contributor License Agreement for you (the sender of this pull request), but were unable to find agreements for all the commit author(s) or Co-authors. If you authored these, maybe you used a different email address in the git commits than was used to sign the CLA (login here to double check)? If these were authored by someone else, then they will need to sign a CLA as well, and confirm that they're okay with these being contributed to Google.
In order to pass this check, please resolve this problem and have the pull request author add another comment and the bot will run again. If the bot doesn't comment, it means it doesn't think anything has changed.

is because is not currently available in current Composer Airflow version (1.9.0)

1. Uses gcs_to_gcs operator from plugins 2. Uses Cloud logging features 3. Dynamically generates tasks based on master csv file

googlebot · 2018-09-01T17:06:15Z

CLAs look good, thanks!

dsdinter · 2018-09-03T14:06:53Z

@tswast can you please review and approve if appropriate?

- Enable export and import multiple Avro files

composer/plugins/__init__.py

tswast · 2018-09-05T00:30:18Z

composer/plugins/gcs_plugin/hooks/gcs_hook.py

+from googleapiclient import errors
+
+
+class GoogleCloudStorageHook(GoogleCloudBaseHook):


I'm a bit confused. Are these files copied from the Airflow repository? If so, could we provide download instructions rather than copying them here?

Correct, both hook and operator are from the Airflow master branch (1.10), I just copied this snapshot version to avoid breaking the code with future changes in the Airflow project.
Let me know if there is a more elegant way (I guess can give download instructions from a specific git commit from the Airflow repo).

OK, I removed these and added instructions to download with wget from 1.10-stable branch.

Please remove the operator as well.

If we cannot, please retain the original license headers and move to a third_party/ directory per go/thirdparty#projects

composer/workflows/bq_copy_across_locations.py

tswast · 2018-09-05T22:22:16Z

composer/workflows/bq_copy_across_locations.py

+import logging
+
+from airflow import models
+from airflow.contrib.operators.bigquery_to_gcs import \


Import modules not classes. from airflow.contrib.operators import bigquery_to_gcs. Likewise for other imports.

tswast · 2018-09-05T22:24:20Z

composer/workflows/bq_copy_across_locations.py

+
+This DAG relies on three Airflow variables
+https://airflow.apache.org/concepts.html#variables
+* master_file_path - CSV file listing source and target tables, including


I suggest table_definition_file_path to avoid use of word "master"

Done - I renamed to "table_list_file_path"

tswast · 2018-09-05T22:25:05Z

composer/workflows/bq_copy_across_locations.py

+# --------------------------------------------------------------------------------
+
+# Instantiates a client
+client = google.cloud.logging.Client()


Is it common to set up logging within DAG parsing? Won't Airflow and/or Composer ensure logs get written to the correct place? Why is this code present?

Code is actually there to test logging against GCP Stackdriver while running in local Airflow environment, I would remove to keep the example clean as it is less relevant with Composer sending logs automatically to Stackdriver.

tswast · 2018-09-05T22:25:53Z

composer/workflows/bq_copy_across_locations.py

+# --------------------------------------------------------------------------------
+
+
+def read_master_file(master_file):


read_table_definition_file

tswast · 2018-09-05T22:27:31Z

composer/workflows/bq_copy_across_locations.py

+    )
+
+    # Get the table list from master file
+    all_records = read_master_file(master_file_path)


@fenglu-g Do you have thoughts on doing dynamic calls to construct a DAG a parse time? I thought this was not a recommended practice.

tswast

Looking good. A few more things to fix up.

Will this need to be updated in a few weeks once Composer supports Airflow 1.10?

tswast · 2018-09-06T22:47:18Z

composer/plugins/gcs_plugin/operators/gcs_to_gcs.py

@@ -0,0 +1,172 @@
+# Copyright 2018 Google LLC


Please remove the operator as well.

If we cannot, please retain the original license headers (we should definitely not replace any license headers with Google copyright) and move to a third_party/ directory per go/thirdparty#projects

tswast · 2018-09-06T22:50:33Z

composer/workflows/bq_copy_across_locations.py

+# --------------------------------------------------------------------------------
+
+import csv
+from datetime import datetime


Import modules, not classes/functions. import datetime.

tswast · 2018-09-06T22:51:54Z

composer/workflows/bq_copy_across_locations_test.py

+    environment. This is a recommended sanity check by the official Airflow
+    docs: https://airflow.incubator.apache.org/tutorial.html#testing
+    """
+    models.Variable.set('master_file_path', 'example_file_path')


Update variable name. Also, the file path should point to bq_copy_eu_to_us_sample.csv, right?

Use __file__ to get the path to this Python file and construct the path from there. For example:

python-docs-samples/appengine/standard/i18n/i18n_utils.py

Lines 180 to 181 in 35fe662

locale_path = os.path.join(

os.path.abspath(os.path.dirname(__file__)), 'locales')

tswast · 2018-09-06T22:52:38Z

composer/workflows/bq_copy_across_locations_test.py

+    """
+    models.Variable.set('master_file_path', 'example_file_path')
+    models.Variable.set('gcs_source_bucket', 'example-project')
+    models.Variable.set('gcs_dest_bucket', 'us-central1-f')


Need to import bq_copy_across_locations so that we can verify that it correctly parses.

…g for Composer DAG as plugin

plugins

…ators

I moved the plugins directory to third-party to avoid nox from running the linter and tests on Apache Airflow code.

tswast · 2018-09-07T23:51:27Z

LGTM. I rearrange the files a little bit and updated the tests in e3adbda

tswast · 2018-09-07T23:51:38Z

@dpebot merge when green

dpebot · 2018-09-07T23:51:39Z

Okay! I'll merge when all statuses are green and all reviewers approve.

googlebot added the cla: no This human has *not* signed the Contributor License Agreement. label Aug 31, 2018

dsdinter added 4 commits September 1, 2018 18:04

Add gcs_to_gcs operator from current Airflow master as a plugin, this

ab87096

is because is not currently available in current Composer Airflow version (1.9.0)

Add gcs_to_gcs operator from current Airflow master as a plugin, this

9fd2eeb

is because is not currently available in current Composer Airflow version (1.9.0)

Author DAG, including master csv file sample and test:

322acd9

1. Uses gcs_to_gcs operator from plugins 2. Uses Cloud logging features 3. Dynamically generates tasks based on master csv file

Fix PEP8 warnings

fac1ed1

googlebot added cla: yes This human has signed the Contributor License Agreement. and removed cla: no This human has *not* signed the Contributor License Agreement. labels Sep 1, 2018

dsdinter added 2 commits September 4, 2018 12:03

Change sample tables to copy, using NYC Taxi trips

b4a5b11

- Replace ":" with valid character for Airflow task

1614048

- Enable export and import multiple Avro files

tswast reviewed Sep 5, 2018

View reviewed changes

dsdinter added 2 commits September 5, 2018 10:07

Add Google LLC Apache 2.0 license header

f306c72

The GCS hook must be downloaded from the Airflow repository

8a0b682

tswast reviewed Sep 5, 2018

View reviewed changes

dsdinter added 7 commits September 6, 2018 13:29

Fix line wrapping in comments

56bb462

Import modules not classes

493915d

Rename config file to "table_list_file_path"

1ee6d6c

Rename master references to table_list

4a3b54c

Remove unnecessary logging code

89b4ccd

Fix lint error

f0e0f78

Refactor master to table_file in code and filename

76d2bfa

tswast reviewed Sep 6, 2018

View reviewed changes

dsdinter added 6 commits September 7, 2018 15:00

Add gcs_to_gcs module from Airflow 1.10 as third party before adaptin…

e869d76

…g for Composer DAG as plugin

Wrap lines in gcs_to_gcs module from Airflow 1.10 and import hook from

b363bbc

plugins

Add notes to install module

a03b0db

Remove gcs_to_gcs module after being moved to third_party folder

ae9f711

Fix lint error

e3afa96

Import modules not classes

d0c9257

dsdinter and others added 6 commits September 7, 2018 15:35

Refactor test and include actual sample table list file

2e0ed85

Fix lint error

58e7554

Fix lint error

9106823

Fix lint error

3f0833f

Add license and instructions in third party folder for hooks and oper…

d1e3f3f

…ators

Fix tests and update CSV parsing for Python 3 compat.

e3adbda

I moved the plugins directory to third-party to avoid nox from running the linter and tests on Apache Airflow code.

tswast approved these changes Sep 7, 2018

View reviewed changes

dpebot added the automerge Merge the pull request once unit tests and other checks pass. label Sep 7, 2018

dpebot self-assigned this Sep 7, 2018

tswast mentioned this pull request Sep 7, 2018

Test Composer samples on Python 3 #1688

Merged

tswast merged commit 16de3b4 into GoogleCloudPlatform:master Sep 8, 2018

		from googleapiclient import errors


		class GoogleCloudStorageHook(GoogleCloudBaseHook):

		# --------------------------------------------------------------------------------


		def read_master_file(master_file):

	locale_path = os.path.join(
	os.path.abspath(os.path.dirname(__file__)), 'locales')

Adding sample python DAG and gcs_to_gcs operator as plugin #1678

Adding sample python DAG and gcs_to_gcs operator as plugin #1678

Uh oh!

Conversation

dsdinter commented Aug 31, 2018

Uh oh!

googlebot commented Aug 31, 2018

Uh oh!

googlebot commented Sep 1, 2018

Uh oh!

dsdinter commented Sep 3, 2018

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tswast left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tswast commented Sep 7, 2018

Uh oh!

tswast commented Sep 7, 2018

Uh oh!

dpebot commented Sep 7, 2018

Uh oh!

Uh oh!