Collection of transforms for the Apache beam python SDK.
Project description
A collection of random transforms for the Apache beam python SDK . Many are simple (or trivial) transforms. The most useful ones are those for reading/writing from/to relational databases.
Installation
- Using pip
pip install beam-nuggets
- From source
git clone git@github.com:mohaseeb/beam-nuggets.git
cd beam-nuggets
pip install .
Usage
Write data to an SQLite table using beam-nugget's relational_db.Write transform.
# write_sqlite.py contents
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from beam_nuggets.io import relational_db
records = [
{'name': 'Jan', 'num': 1},
{'name': 'Feb', 'num': 2}
]
source_config = relational_db.SourceConfiguration(
drivername='sqlite',
database='/tmp/months_db.sqlite',
create_if_missing=True # create the database if not there
)
table_config = relational_db.TableConfiguration(
name='months',
create_if_missing=True, # automatically create the table if not there
primary_key_columns=['num'] # and use 'num' column as a primary key
)
with beam.Pipeline(options=PipelineOptions()) as p: # Will use local runner
months = p | "Reading month records" >> beam.Create(records)
months | 'Writing to DB' >> relational_db.Write(
source_config=source_config,
table_config=table_config
)
Execute the pipeline
python write_sqlite.py
Examine the contents
sqlite3 /tmp/months_db.sqlite 'select * from months'
# output:
# 1.0|Jan
# 2.0|Feb
To write the same data to a PostgreSQL table instead, just create a suitable relational_db.SourceConfiguration as follows.
source_config = relational_db.SourceConfiguration(
drivername='postgresql',
host='localhost',
port=5432,
username='postgres',
password='password',
database='calendar',
create_if_missing=True # create the database if not there
)
An example showing how you can use beam-nugget's relational_db.Read transform to read from a PostgreSQL database table.
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from beam_nuggets.io import relational_db
with beam.Pipeline(options=PipelineOptions()) as p:
source_config = relational_db.SourceConfiguration(
drivername='postgresql',
host='localhost',
port=5432,
username='postgres',
password='password',
database='calendar',
)
records = p | "Reading records from db" >> relational_db.Read(
source_config=source_config,
table_name='months',
)
records | 'Writing to stdout' >> beam.Map(print)
See here for more examples.
Supported transforms
IO
- relational_db.Read for reading from relational database tables.
- relational_db.Write
for writing to relational database tables.
Above transforms uses SqlAlchemy to communicate with the database, and hence they can read from and write to all relational databases supported by SqlAlchemy. The transforms are tested against PostgreSQL, MySQL and SQLite.
- csvio.Read for reading CSV files.
Others
- SelectFromNestedDict Selects a subset from records formed of nested dictionaries.
- ParseJson
- AssignUniqueId
Documentation
See here.
Development
- Install
git clone git@github.com:mohaseeb/beam-nuggets.git
cd beam-nuggets
export BEAM_NUGGETS_ROOT=`pwd`
pip install -e .[dev]
- Make changes on dedicated dev branches
- Run tests
cd $BEAM_NUGGETS_ROOT
python -m unittest discover -v
- Generate docs
cd $BEAM_NUGGETS_ROOT
docs/generate_docs.sh
- Create a PR against master.
- After merging the accepted PR and updating the local master, upload a new build to pypi.
cd $BEAM_NUGGETS_ROOT
scripts/build_test_deploy.sh
Backlog
- versioned docs?
- Summarize the investigation of using Source/Sink Vs ParDo(and GroupBy) for IO
- instruct how to use other DB drivers supported by sqlalchemy
- Example how to run on GCP
- Sql queries support in relational_db.Read
- more nuggets: WriteToCsv
- integration tests
- DB transforms failures handling on IO transforms
- more nuggets: Elasticsearch, Mongo
- WriteToRelationalDB, logging
Licence
MIT
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for beam_nuggets-0.10.0-py2-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 607163428647ca8641ce6479910a3f84a3375990c7f51d8cfead62e9be5de23f |
|
MD5 | 3de6100939f85f38a55f1d40a61a79b7 |
|
BLAKE2b-256 | 01b33f039202a0e010a05bbd8c7dc19b39a7485a44270419134a2d0aed27aa40 |