#NEWLINE# def run( argv = None ) : #NEWLINE# #INDENT# 'Run the workflow.' #NEWLINE# parser = argparse . ArgumentParser ( ) #NEWLINE# parser . add_argument ( '--output' , required = True , help = 'Output BigQuery table for results specified as: PROJECT:DATASET.TABLE or DATASET.TABLE.' ) #NEWLINE# ( known_args , pipeline_args ) = parser . parse_known_args ( argv ) #NEWLINE# p = df . Pipeline ( argv = pipeline_args ) #NEWLINE# from google.cloud.dataflow.internal.clients import bigquery #NEWLINE# table_schema = bigquery . TableSchema ( ) #NEWLINE# kind_schema = bigquery . TableFieldSchema ( ) #NEWLINE# kind_schema . name = 'kind' #NEWLINE# kind_schema . type = 'string' #NEWLINE# kind_schema . mode = 'nullable' #NEWLINE# table_schema . fields . append ( kind_schema ) #NEWLINE# full_name_schema = bigquery . TableFieldSchema ( ) #NEWLINE# full_name_schema . name = 'fullName' #NEWLINE# full_name_schema . type = 'string' #NEWLINE# full_name_schema . mode = 'required' #NEWLINE# table_schema . fields . append ( full_name_schema ) #NEWLINE# age_schema = bigquery . TableFieldSchema ( ) #NEWLINE# age_schema . name = 'age' #NEWLINE# age_schema . type = 'integer' #NEWLINE# age_schema . mode = 'nullable' #NEWLINE# table_schema . fields . append ( age_schema ) #NEWLINE# gender_schema = bigquery . TableFieldSchema ( ) #NEWLINE# gender_schema . name = 'gender' #NEWLINE# gender_schema . type = 'string' #NEWLINE# gender_schema . mode = 'nullable' #NEWLINE# table_schema . fields . append ( gender_schema ) #NEWLINE# phone_number_schema = bigquery . TableFieldSchema ( ) #NEWLINE# phone_number_schema . name = 'phoneNumber' #NEWLINE# phone_number_schema . type = 'record' #NEWLINE# phone_number_schema . mode = 'nullable' #NEWLINE# area_code = bigquery . TableFieldSchema ( ) #NEWLINE# area_code . name = 'areaCode' #NEWLINE# area_code . type = 'integer' #NEWLINE# area_code . mode = 'nullable' #NEWLINE# phone_number_schema . fields . append ( area_code ) #NEWLINE# number = bigquery . TableFieldSchema ( ) #NEWLINE# number . name = 'number' #NEWLINE# number . type = 'integer' #NEWLINE# number . mode = 'nullable' #NEWLINE# phone_number_schema . fields . append ( number ) #NEWLINE# table_schema . fields . append ( phone_number_schema ) #NEWLINE# children_schema = bigquery . TableFieldSchema ( ) #NEWLINE# children_schema . name = 'children' #NEWLINE# children_schema . type = 'string' #NEWLINE# children_schema . mode = 'repeated' #NEWLINE# table_schema . fields . append ( children_schema ) #NEWLINE# def create_random_record( record_id ) : #NEWLINE# #INDENT# return { #NEWLINE# #INDENT# 'kind' : ( 'kind' + record_id ) , #NEWLINE# 'fullName' : ( 'fullName' + record_id ) , #NEWLINE# 'age' : ( int ( record_id ) * 10 ) , #NEWLINE# 'gender' : 'male' , #NEWLINE# 'phoneNumber' : { #NEWLINE# #INDENT# 'areaCode' : ( int ( record_id ) * 100 ) , #NEWLINE# 'number' : ( int ( record_id ) * 100000 ) , #NEWLINE# #UNINDENT# } , #NEWLINE# 'children' : [ ( ( 'child' + record_id ) + '1' ) , ( ( 'child' + record_id ) + '2' ) , ( ( 'child' + record_id ) + '3' ) ] , #NEWLINE# #UNINDENT# } #NEWLINE# #UNINDENT# record_ids = ( p | df . Create ( 'CreateIDs' , [ '1' , '2' , '3' , '4' , '5' ] ) ) #NEWLINE# records = ( record_ids | df . Map ( 'CreateRecords' , create_random_record ) ) #NEWLINE# ( records | df . io . Write ( 'write' , df . io . BigQuerySink ( known_args . output , schema = table_schema , create_disposition = df . io . BigQueryDisposition . CREATE_IF_NEEDED , write_disposition = df . io . BigQueryDisposition . WRITE_TRUNCATE ) ) ) #NEWLINE# p . run ( )