--Easy data processing ――It will balance the workload of the worker without permission. ――It works on GCE so you can do anything --Datalab (Jupyter) You can deploy from above!
** Simply put, a super cool version of MapReduce **
git clone https://github.com/hayatoy/dataflow-tutorial.git
--Google Cloud Platform billing settings
--Enable Dataflow API
-Create a GCS Bucket
-Create a dataset called testdataset in BigQuery
--Please change the Project name.
Replacing PROJECTID is faster.
That's it!
import apache_beam as beam
Specify the job name, project name, and temporary file location.
options = beam.utils.pipeline_options.PipelineOptions()
gcloud_options = options.view_as(
beam.utils.pipeline_options.GoogleCloudOptions)
gcloud_options.job_name = 'dataflow-tutorial1'
gcloud_options.project = 'PROJECTID'
gcloud_options.staging_location = 'gs://PROJECTID/staging'
gcloud_options.temp_location = 'gs://PROJECTID/temp'
Set the maximum number of workers, machine type, etc. The disk size of the worker is ** large, 250GB (Batch) and 420GB (Streaming) by default **, so it is recommended to specify the required size here.
worker_options = options.view_as(beam.utils.pipeline_options.WorkerOptions)
worker_options.disk_size_gb = 20
worker_options.max_num_workers = 2
# worker_options.num_workers = 2
# worker_options.machine_type = 'n1-standard-8'
# worker_options.zone = 'asia-northeast1-a'
-** DirectRunner: ** Run on local machine -** DataflowRunner: ** Run on Dataflow
options.view_as(beam.utils.pipeline_options.StandardOptions).runner = 'DirectRunner'
# options.view_as(beam.utils.pipeline_options.StandardOptions).runner = 'DataflowRunner'
Just read the file from GCS and write its contents to GCS
+----------------+
| |
| Read GCS File |
| |
+-------+--------+
|
v
+-------+--------+
| |
| Write GCS File |
| |
+----------------+
p1 = beam.Pipeline(options=options)
(p1 | 'read' >> beam.io.ReadFromText('gs://dataflow-samples/shakespeare/kinglear.txt')
| 'write' >> beam.io.WriteToText('gs://PROJECTID/test.txt', num_shards=1)
)
p1.run().wait_until_finish()
Just read the data from BigQuery and write its contents to GCS BigQuery dataset is below https://bigquery.cloud.google.com/table/bigquery-public-data:samples.shakespeare
+----------------+
| |
| Read BigQuery |
| |
+-------+--------+
|
v
+-------+--------+
| |
| Write GCS File |
| |
+----------------+
p2 = beam.Pipeline(options=options)
query = 'SELECT * FROM [bigquery-public-data:samples.shakespeare] LIMIT 10'
(p2 | 'read' >> beam.io.Read(beam.io.BigQuerySource(project='PROJECTID', use_standard_sql=False, query=query))
| 'write' >> beam.io.WriteToText('gs://PROJECTID/test2.txt', num_shards=1)
)
p2.run().wait_until_finish()
Read data from BigQuery and write data to BigQuery
+----------------+
| |
| Read BigQuery |
| |
+-------+--------+
|
v
+-------+--------+
| |
| Write BigQuery |
| |
+----------------+
p3 = beam.Pipeline(options=options)
#Note: Create a dataset
query = 'SELECT * FROM [bigquery-public-data:samples.shakespeare] LIMIT 10'
(p3 | 'read' >> beam.io.Read(beam.io.BigQuerySource(project='PROJECTID', use_standard_sql=False, query=query))
| 'write' >> beam.io.Write(beam.io.BigQuerySink(
'testdataset.testtable1',
schema='corpus_date:INTEGER, corpus:STRING, word:STRING, word_count:INTEGER',
create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE))
)
p3.run().wait_until_finish()
--Read data from BigQuery --Process the data --Write to BigQuery
+----------------+
| |
| Read BigQuery |
| |
+-------+--------+
|
v
+-------+--------+
| |
| Modify Element |
| |
+----------------+
|
v
+-------+--------+
| |
| Write BigQuery |
| |
+----------------+
def modify_data1(element):
# beam.Map is used to output one line for one line of input
# element = {u'corpus_date': 0, u'corpus': u'sonnets', u'word': u'LVII', u'word_count': 1}
corpus_upper = element['corpus'].upper()
word_len = len(element['word'])
return {'corpus_upper': corpus_upper,
'word_len': word_len
}
p4 = beam.Pipeline(options=options)
query = 'SELECT * FROM [bigquery-public-data:samples.shakespeare] LIMIT 10'
(p4 | 'read' >> beam.io.Read(beam.io.BigQuerySource(project='PROJECTID', use_standard_sql=False, query=query))
| 'modify' >> beam.Map(modify_data1)
| 'write' >> beam.io.Write(beam.io.BigQuerySink(
'testdataset.testtable2',
schema='corpus_upper:STRING, word_len:INTEGER',
create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE))
)
p4.run().wait_until_finish()
Example of splitting a branch
+----------------+
| |
| Read BigQuery |
| |
+-------+--------+
|
+---------------------+
| |
+-------v--------+ +-------v--------+
| | | |
| Modify Element | | Modify Element |
| | | |
+-------+--------+ +-------+--------+
| |
+---------------------+
|
+-------v--------+
| |
| Flatten |
| |
+-------+--------+
|
|
+-------v--------+
| |
| Save BigQuery |
| |
+----------------+
def modify1(element):
# element = {u'corpus_date': 0, u'corpus': u'sonnets', u'word': u'LVII', u'word_count': 1}
word_count = len(element['corpus'])
count_type = 'corpus only'
return {'word_count': word_count,
'count_type': count_type
}
def modify2(element):
# element = {u'corpus_date': 0, u'corpus': u'sonnets', u'word': u'LVII', u'word_count': 1}
word_count = len(element['word'])
count_type = 'word only'
return {'word_count': word_count,
'count_type': count_type
}
p5 = beam.Pipeline(options=options)
query = 'SELECT * FROM [bigquery-public-data:samples.shakespeare] LIMIT 10'
query_results = p5 | 'read' >> beam.io.Read(beam.io.BigQuerySource(
project='PROJECTID', use_standard_sql=False, query=query))
#Pass BigQuery results to two branches
branch1 = query_results | 'modify1' >> beam.Map(modify1)
branch2 = query_results | 'modify2' >> beam.Map(modify2)
#Flatten the results from the branch
((branch1, branch2) | beam.Flatten()
| 'write' >> beam.io.Write(beam.io.BigQuerySink(
'testdataset.testtable3',
schema='word_count:INTEGER, count_type:STRING',
create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE))
)
p5.run().wait_until_finish()
Use ** Group by **
def modify_data2(kvpair):
#groupby passes a tuple of keys and a list of data with those keys
# kvpair = (u'word only', [4, 4, 6, 6, 7, 7, 7, 7, 8, 9])
return {'count_type': kvpair[0],
'sum': sum(kvpair[1])
}
p6 = beam.Pipeline(options=options)
query = 'SELECT * FROM [PROJECTID:testdataset.testtable3] LIMIT 20'
(p6 | 'read' >> beam.io.Read(beam.io.BigQuerySource(project='PROJECTID', use_standard_sql=False, query=query))
| 'pair' >> beam.Map(lambda x: (x['count_type'], x['word_count']))
| "groupby" >> beam.GroupByKey()
| 'modify' >> beam.Map(modify_data2)
| 'write' >> beam.io.Write(beam.io.BigQuerySink(
'testdataset.testtable4',
schema='count_type:STRING, sum:INTEGER',
create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE))
)
p6.run().wait_until_finish()
Separate the section of ** GroupBy ** with ** Window **
def assign_timevalue(v):
#Add a time stamp to the pcollection data
#The latter window is divided based on this time stamp
#Here, time stamps are put in with random numbers.
import apache_beam.transforms.window as window
import random
import time
return window.TimestampedValue(v, int(time.time()) + random.randint(0, 1))
def modify_data3(kvpair):
#groupby passes a tuple of keys and a list of data with those keys
#Since it is divided by window, the number of data is small
# kvpair = (u'word only', [4, 4, 6, 6, 7])
return {'count_type': kvpair[0],
'sum': sum(kvpair[1])
}
p7 = beam.Pipeline(options=options)
query = 'SELECT * FROM [PROJECTID:testdataset.testtable3] LIMIT 20'
(p7 | 'read' >> beam.io.Read(beam.io.BigQuerySource(project='PROJECTID', use_standard_sql=False, query=query))
| "assign tv" >> beam.Map(assign_timevalue)
| 'window' >> beam.WindowInto(beam.window.FixedWindows(1))
| 'pair' >> beam.Map(lambda x: (x['count_type'], x['word_count']))
| "groupby" >> beam.GroupByKey()
| 'modify' >> beam.Map(modify_data3)
| 'write' >> beam.io.Write(beam.io.BigQuerySink(
'testdataset.testtable5',
schema='count_type:STRING, sum:INTEGER',
create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE))
)
p7.run().wait_until_finish()