I am currently working on a ETL Dataflow job (using the Apache Beam Python SDK) which queries data from CloudSQL (with psycopg2 and a custom ParDo) and writes it to BigQuery. My goal is to create a Dataflow template which I can start from a AppEngine using a Cron job.
I have a version which works locally using the DirectRunner. For that I use the CloudSQL (Postgres) proxy client so that I can connect to the database on 127.0.0.1 .
When using the DataflowRunner with custom commands to start the proxy within a setup.py script, the job won’t execute. It stucks with repeating this log-message:
Setting node annotation to enable volume controller attach/detach
A part of my setup.py looks the following:
JavaScript
x
34
34
1
CUSTOM_COMMANDS = [2
['echo', 'Custom command worked!'],3
['wget', 'https://dl.google.com/cloudsql/cloud_sql_proxy.linux.amd64', '-O', 'cloud_sql_proxy'],4
['echo', 'Proxy downloaded'],5
['chmod', '+x', 'cloud_sql_proxy']]6
7
class CustomCommands(setuptools.Command):8
"""A setuptools Command class able to run arbitrary commands."""9
10
def initialize_options(self):11
pass12
13
def finalize_options(self):14
pass15
16
def RunCustomCommand(self, command_list):17
print('Running command: %s' % command_list)18
logging.info("Running custom commands")19
p = subprocess.Popen(20
command_list,21
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)22
# Can use communicate(input='yn'.encode()) if the command run requires23
# some confirmation.24
stdout_data, _ = p.communicate()25
print('Command output: %s' % stdout_data)26
if p.returncode != 0:27
raise RuntimeError(28
'Command %s failed: exit code: %s' % (command_list, p.returncode))29
30
def run(self):31
for command in CUSTOM_COMMANDS:32
self.RunCustomCommand(command)33
subprocess.Popen(['./cloud_sql_proxy', '-instances=bi-test-1:europe-west1:test-animal=tcp:5432'])34
I added the last line as separate subprocess.Popen() within run() after reading this issue on Github from sthomp and this discussion on Stackoverflo. I also tried to play around with some parameters of subprocess.Popen.
Another mentioned solution from brodin was to allow access from every IP address and to connect via username and password. In my understanding he does not claim this as best practice.
Thank you in advance for you help.
!!! Workaround solution at bottom of this post !!!
Update – Logfiles
These are the logs on error level which occur during a job:
JavaScript
1
28
28
1
E EXT4-fs (dm-0): couldn't mount as ext3 due to feature incompatibilities 2
E Image garbage collection failed once. Stats initialization may not have completed yet: unable to find data for container / 3
E Failed to check if disk space is available for the runtime: failed to get fs info for "runtime": unable to find data for container / 4
E Failed to check if disk space is available on the root partition: failed to get fs info for "root": unable to find data for container / 5
E [ContainerManager]: Fail to get rootfs information unable to find data for container / 6
E Could not find capacity information for resource storage.kubernetes.io/scratch 7
E debconf: delaying package configuration, since apt-utils is not installed 8
E % Total % Received % Xferd Average Speed Time Time Time Current 9
E Dload Upload Total Spent Left Speed 10
E 11
0 0 0 0 0 0 0 0 --:--:-- --:--:-- --:--:-- 012
100 3698 100 3698 0 0 25674 0 --:--:-- --:--:-- --:--:-- 25860 13
14
15
16
#-- HERE IS WHEN setup.py FOR MY JOB IS EXECUTED ---17
18
E debconf: delaying package configuration, since apt-utils is not installed 19
E insserv: warning: current start runlevel(s) (empty) of script `stackdriver-extractor' overrides LSB defaults (2 3 4 5). 20
E insserv: warning: current stop runlevel(s) (0 1 2 3 4 5 6) of script `stackdriver-extractor' overrides LSB defaults (0 1 6). 21
E option = Interval; value = 60.000000; 22
E option = FQDNLookup; value = false; 23
E Created new plugin context. 24
E option = PIDFile; value = /var/run/stackdriver-agent.pid; 25
E option = Interval; value = 60.000000; 26
E option = FQDNLookup; value = false; 27
E Created new plugin context. 28
Here you can find are all logs after the start of my custom setup.py (log-level: any; all logs):
Update logfiles 2
Job logs (I manually canceled the job after not stucking for a while):
JavaScript
1
48
48
1
2018-06-08 (08:02:20) Autoscaling is enabled for job 2018-06-07_23_02_20-5917188751755240698. The number of workers will b2
2018-06-08 (08:02:20) Autoscaling was automatically enabled for job 2018-06-07_23_02_20-5917188751755240698.3
2018-06-08 (08:02:24) Checking required Cloud APIs are enabled.4
2018-06-08 (08:02:24) Checking permissions granted to controller Service Account.5
2018-06-08 (08:02:25) Worker configuration: n1-standard-1 in europe-west1-b.6
2018-06-08 (08:02:25) Expanding CoGroupByKey operations into optimizable parts.7
2018-06-08 (08:02:25) Combiner lifting skipped for step Save new watermarks/Write/WriteImpl/GroupByKey: GroupByKey not fol8
2018-06-08 (08:02:25) Combiner lifting skipped for step Group watermarks: GroupByKey not followed by a combiner.9
2018-06-08 (08:02:25) Expanding GroupByKey operations into optimizable parts.10
2018-06-08 (08:02:26) Lifting ValueCombiningMappingFns into MergeBucketsMappingFns11
2018-06-08 (08:02:26) Annotating graph with Autotuner information.12
2018-06-08 (08:02:26) Fusing adjacent ParDo, Read, Write, and Flatten operations13
2018-06-08 (08:02:26) Fusing consumer Get rows from CloudSQL tables into Begin pipeline with watermarks/Read14
2018-06-08 (08:02:26) Fusing consumer Group watermarks/Write into Group watermarks/Reify15
2018-06-08 (08:02:26) Fusing consumer Group watermarks/GroupByWindow into Group watermarks/Read16
2018-06-08 (08:02:26) Fusing consumer Save new watermarks/Write/WriteImpl/WriteBundles/WriteBundles into Save new watermar17
2018-06-08 (08:02:26) Fusing consumer Save new watermarks/Write/WriteImpl/GroupByKey/GroupByWindow into Save new watermark18
2018-06-08 (08:02:26) Fusing consumer Save new watermarks/Write/WriteImpl/GroupByKey/Reify into Save new watermarks/Write/19
2018-06-08 (08:02:26) Fusing consumer Save new watermarks/Write/WriteImpl/GroupByKey/Write into Save new watermarks/Write/20
2018-06-08 (08:02:26) Fusing consumer Write to BQ into Get rows from CloudSQL tables21
2018-06-08 (08:02:26) Fusing consumer Group watermarks/Reify into Write to BQ22
2018-06-08 (08:02:26) Fusing consumer Save new watermarks/Write/WriteImpl/Map(<lambda at iobase.py:926>) into Convert dict23
2018-06-08 (08:02:26) Fusing consumer Save new watermarks/Write/WriteImpl/WindowInto(WindowIntoFn) into Save new watermark24
2018-06-08 (08:02:26) Fusing consumer Convert dictionary list to single dictionary and json into Remove "watermark" label25
2018-06-08 (08:02:26) Fusing consumer Remove "watermark" label into Group watermarks/GroupByWindow26
2018-06-08 (08:02:26) Fusing consumer Save new watermarks/Write/WriteImpl/InitializeWrite into Save new watermarks/Write/W27
2018-06-08 (08:02:26) Workflow config is missing a default resource spec.28
2018-06-08 (08:02:26) Adding StepResource setup and teardown to workflow graph.29
2018-06-08 (08:02:26) Adding workflow start and stop steps.30
2018-06-08 (08:02:26) Assigning stage ids.31
2018-06-08 (08:02:26) Executing wait step start2532
2018-06-08 (08:02:26) Executing operation Save new watermarks/Write/WriteImpl/DoOnce/Read+Save new watermarks/Write/WriteI33
2018-06-08 (08:02:26) Executing operation Save new watermarks/Write/WriteImpl/GroupByKey/Create34
2018-06-08 (08:02:26) Starting worker pool setup.35
2018-06-08 (08:02:26) Executing operation Group watermarks/Create36
2018-06-08 (08:02:26) Starting 1 workers in europe-west1-b37
2018-06-08 (08:02:27) Value "Group watermarks/Session" materialized.38
2018-06-08 (08:02:27) Value "Save new watermarks/Write/WriteImpl/GroupByKey/Session" materialized.39
2018-06-08 (08:02:27) Executing operation Begin pipeline with watermarks/Read+Get rows from CloudSQL tables+Write to BQ+Gr40
2018-06-08 (08:02:36) Autoscaling: Raised the number of workers to 0 based on the rate of progress in the currently runnin41
2018-06-08 (08:02:46) Autoscaling: Raised the number of workers to 1 based on the rate of progress in the currently runnin42
2018-06-08 (08:03:05) Workers have started successfully.43
2018-06-08 (08:11:37) Cancel request is committed for workflow job: 2018-06-07_23_02_20-5917188751755240698.44
2018-06-08 (08:11:38) Cleaning up.45
2018-06-08 (08:11:38) Starting worker pool teardown.46
2018-06-08 (08:11:38) Stopping worker pool47
2018-06-08 (08:12:30) Autoscaling: Reduced the number of workers to 0 based on the rate of progress in the currently runni48
Stack Traces:
JavaScript
1
2
1
No errors have been received in this time period.2
Update: Workaround Solution can be found in my answer below
Advertisement
Answer
Workaround Solution:
I finally found a workaround. I took the idea to connect via the public IP of the CloudSQL instance. For that you needed to allow connections to your CloudSQL instance from every IP:
- Go to the overview page of your CloudSQL instance in GCP
- Click on the
Authorizationtab - Click on
Add networkand add0.0.0.0/0(!! this will allow every IP address to connect to your instance !!)
To add security to the process, I used SSL keys and only allowed SSL connections to the instance:
- Click on
SSLtab - Click on
Create a new certificateto create a SSL certificate for your server - Click on
Create a client certificateto create a SSL certificate for you client - Click on
Allow only SSL connectionsto reject all none SSL connection attempts
After that I stored the certificates in a Google Cloud Storage bucket and load them before connecting within the Dataflow job, i.e.:
JavaScript
1
50
50
1
import psycopg22
import psycopg2.extensions3
import os4
import stat5
from google.cloud import storage6
7
# Function to wait for open connection when processing parallel8
def wait(conn):9
while 1:10
state = conn.poll()11
if state == psycopg2.extensions.POLL_OK:12
break13
elif state == psycopg2.extensions.POLL_WRITE:14
pass15
select.select([], [conn.fileno()], [])16
elif state == psycopg2.extensions.POLL_READ:17
pass18
select.select([conn.fileno()], [], [])19
else:20
raise psycopg2.OperationalError("poll() returned %s" % state)21
22
# Function which returns a connection which can be used for queries23
def connect_to_db(host, hostaddr, dbname, user, password, sslmode = 'verify-full'):24
25
# Get keys from GCS26
client = storage.Client()27
28
bucket = client.get_bucket(<YOUR_BUCKET_NAME>)29
30
bucket.get_blob('PATH_TO/server-ca.pem').download_to_filename('server-ca.pem')31
bucket.get_blob('PATH_TO/client-key.pem').download_to_filename('client-key.pem')32
os.chmod("client-key.pem", stat.S_IRWXU)33
bucket.get_blob('PATH_TO/client-cert.pem').download_to_filename('client-cert.pem')34
35
sslrootcert = 'server-ca.pem'36
sslkey = 'client-key.pem'37
sslcert = 'client-cert.pem'38
39
con = psycopg2.connect(40
host = host,41
hostaddr = hostaddr,42
dbname = dbname,43
user = user,44
password = password,45
sslmode=sslmode,46
sslrootcert = sslrootcert,47
sslcert = sslcert,48
sslkey = sslkey)49
return con50
I then use these functions in a custom ParDo to perform queries.
Minimal example:
JavaScript
1
44
44
1
import apache_beam as beam2
3
class ReadSQLTableNames(beam.DoFn):4
'''5
parDo class to get all table names of a given cloudSQL database.6
It will return each table name.7
'''8
def __init__(self, host, hostaddr, dbname, username, password):9
super(ReadSQLTableNames, self).__init__()10
self.host = host11
self.hostaddr = hostaddr12
self.dbname = dbname13
self.username = username14
self.password = password15
16
def process(self, element):17
18
# Connect do database19
con = connect_to_db(host = self.host,20
hostaddr = self.hostaddr,21
dbname = self.dbname,22
user = self.username,23
password = self.password)24
# Wait for free connection25
wait_select(con)26
# Create cursor to query data27
cur = con.cursor(cursor_factory=RealDictCursor)28
29
# Get all table names30
cur.execute(31
"""32
SELECT33
tablename as table34
FROM pg_tables35
WHERE schemaname = 'public'36
"""37
)38
table_names = cur.fetchall()39
40
cur.close()41
con.close()42
for table_name in table_names:43
yield table_name["table"]44
A part of the pipeline then could look like this:
JavaScript
1
11
11
1
# Current workaround to query all tables: 2
# Create a dummy initiator PCollection with one element3
init = p |'Begin pipeline with initiator' >> beam.Create(['All tables initializer'])4
5
tables = init |'Get table names' >> beam.ParDo(ReadSQLTableNames(6
host = known_args.host,7
hostaddr = known_args.hostaddr,8
dbname = known_args.db_name,9
username = known_args.user,10
password = known_args.password))11
I hope this solution helps others with similar problems