beam-nuggets

What it is and what it does

beam-nuggets is a collection of custom transforms that extend Apache Beam's Python SDK with connectors for relational databases, Kafka, and CSV files. It wraps SQLAlchemy to provide ReadFromDB and Write transforms that work with any SQLAlchemy-supported database (PostgreSQL, MySQL, SQLite tested), plus KafkaConsume and KafkaProduce for Kafka integration and CSV reading. It also includes utility transforms for parsing JSON, selecting from nested dictionaries, and assigning unique IDs.

The package is most useful for data pipelines that need to ingest from or load into SQL databases or Kafka topics within a Beam pipeline. It abstracts away the boilerplate of configuring database connections and serialization, letting you focus on pipeline logic. However, the project is dormant—last updated in September 2021—so it may not work with recent versions of Beam, database drivers, or Kafka clients without manual fixes.

Use it for:

• Read records from a PostgreSQL or MySQL table and process them in a Beam pipeline.
• Write transformed data from a Beam pipeline directly into a relational database table.
• Consume messages from a Kafka topic and process them in a Beam pipeline.
• Produce processed records to a Kafka topic from a Beam pipeline.
• Parse CSV files or JSON data as part of a Beam data processing workflow.

Worth the install?

AI-flagged interpretation of the facts on this page — verify before relying

Provides Apache Beam transforms for reading and writing relational databases, consuming and producing Kafka messages, and parsing CSV files and JSON data.

Yes, if you need Beam-to-database or Beam-to-Kafka connectors and can tolerate dormant maintenance. The package has low install friction and no known vulnerabilities, but verify compatibility with your Beam and driver versions first. If you require active support or need to integrate with very recent Beam or database driver releases, consider forking or maintaining a local patch.

Install

beam-nuggets on PyPI

pip install beam-nuggets

uv add beam-nuggets

poetry add beam-nuggets

Installing beam-nuggets

Before you install

Low friction installation with a pure-Python wheel. Maintenance is dormant—last release was 2021-09-12 and no commits since 2023-12-07—so expect no active bug fixes or updates for newer Beam or database driver versions.

License in practice

MIT license (permissive) allows commercial use, modification, and distribution with minimal restrictions.

Quickstart

pip install beam-nuggets

import apache_beam as beam
from beam_nuggets.io import relational_db

source_config = relational_db.SourceConfiguration(
    drivername='sqlite',
    database='/tmp/test.sqlite'
)
table_config = relational_db.TableConfiguration(name='data')

with beam.Pipeline() as p:
    p | beam.Create([{'id': 1}]) | relational_db.Write(
        source_config=source_config,
        table_config=table_config
    )

Requires Apache Beam to be installed; database drivers (pg8000 for PostgreSQL, PyMySQL for MySQL) must be available for the target database.

Verify before relying

• Compatibility with recent Apache Beam versions (last tested against unknown version as of 2021).
• Whether Kafka transforms work with modern kafka-python API and broker versions.
• Support for Python versions beyond 3.x (exact minor versions unspecified in metadata).

Package facts

Evidence: beam_nuggets-0.18.1-py3-none-any.whl

Tags

More Distributed Computing packages