beam-postgres

Light IO transforms for Postgres read/write in Apache Beam pipelines.

Goal

The project aims to provide highly performant and customizable transforms and is not intended to support many different SQL database engines.

Features

• ReadAllFromPostgres, ReadFromPostgres`` and WriteToPostgres` transforms
• Records can be mapped to tuples, dictionaries or dataclasses
• Reads and writes are in configurable batches

Usage

Printing data from the database table:

importapache_beamasbeamfrompsycopg.rowsimportdict_rowfrombeam_postgres.ioimportReadAllFromPostgreswithbeam.Pipeline() asp:
data=p|"Reading example records from database">>ReadAllFromPostgres(
"host=localhost dbname=examples user=postgres password=postgres",
"select id, data from source",
dict_row,
)
data|"Writing to stdout">>beam.Map(print)

Writing data to the database table:

fromdataclassesimportdataclassimportapache_beamasbeamfromapache_beam.options.pipeline_optionsimportPipelineOptionsfrombeam_postgres.ioimportWriteToPostgres@dataclassclassExample:
data: strwithbeam.Pipeline(options=PipelineOptions()) asp:
data=p|"Reading example records">>beam.Create(
[
Example("example1"),
Example("example2"),
]
)
data|"Writing example records to database">>WriteToPostgres(
"host=localhost dbname=examples user=postgres password=postgres",
"insert into sink (data) values (%(data)s)",
)

See here for more examples.

Reading in batches

There may be situations when you have so much data that it will not fit into the memory - then you want to read your table data in batches. You can see an example code here (the code reads records in a batches of 1).