in courses/understanding_spanner/dataflow/spanner-to-bq.py [0:0]
def main(argv=None, save_main_session=True):
"""Main entry point."""
projectid = os.environ.get('GOOGLE_CLOUD_PROJECT')
parser = argparse.ArgumentParser()
parser.add_argument(
'--instance',
dest='instance',
default='test-spanner-instance',
help='Spanner instance ID.')
parser.add_argument(
'--database',
dest='database',
default = 'pets-db',
help='Spanner database.')
known_args, pipeline_args = parser.parse_known_args(argv)
pipeline_options = PipelineOptions(pipeline_args)
pipeline_options.view_as(SetupOptions).save_main_session = save_main_session
with beam.Pipeline(options=pipeline_options) as p:
owner_pets = p | ReadFromSpanner(
project_id=projectid,
instance_id=known_args.instance,
database_id=known_args.database,
row_type=PetRow,
sql = "SELECT OwnerID, PetName, PetType, Breed FROM Pets"
).with_output_types(PetRow)
( owner_pets | beam.Map(lambda x : x._asdict())
| beam.io.WriteToBigQuery('Pets', dataset = 'petsdb', project = projectid, method = 'STREAMING_INSERTS')
)
owner_pets | beam.Map(print)