Organize ETL Processes into Pipes
Meerschaum Pipes are parametrized ETL processes that are tagged and organized into hierarchies to make scaling up a breeze.
Robust Plugin System
Plugins make it easy to ingest any data source, add functionality to Meerschaum, and organize your utility scripts.
Clean Connector Management
Define your connectors at any level: through the CLI, in your environment, or dynamically.
```bash ### You can follow an interactive wizard. mrsm bootstrap connector ### Or define connectors in your environment. export MRSM_SQL_BAZ='postgresql://foo:bar@localhost:5432/baz' export MRSM_MONGODB_LOCAL='{ "uri": "mongodb://localhost:27017", "database": "meerschaum" }' ``` ```python ### Or you can build connectors on-the-fly in code. import meerschaum as mrsm conn = mrsm.get_connector('sql:demo', flavor='sqlite', database='/tmp/demo.db') ```Design, Develop, and Deploy with Meerschaum Compose
The compose workflow allows you to iterate with and version-control your pipes, making collaboration and maintainability much smoother.
Performant SQL Transformations
SQL pipes with the same instance and source connector are synced in-place β deltas are resolved entirely through SQL and nothing is loaded into RAM.
Concurrency at its Finest
Maximize your throughput by syncing multiple pipes in parallel.
Elegant Chunking
Calm your out-of-memory fears with automatic, parallelized chunking.
Simple-Yet-Powerful API
Want to use Meerschaum in your code? Check out the package documentation! ```python import meerschaum as mrsm conn = mrsm.get_connector("sql:demo", uri="sqlite:////tmp/demo.db") pipe = mrsm.Pipe( 'foo', 'bar', instance = conn, columns = {'datetime': 'dt', 'id': 'id'}, dtypes = {'attrs': 'json'}, ) docs = [ {'dt': '2023-01-01', 'id': 1, 'val': 123.4}, {'dt': '2023-01-01', 'id': 2, 'val': 567.8}, ] pipe.sync(docs) docs = [ {'dt': '2023-01-01', 'id': 1, 'attrs': {'foo': 'bar'}}, ] pipe.sync(docs) df = pipe.get_data(params={'id': [1, 2]}) print(df) # dt id val attrs # 0 2023-01-01 1 123.4 {'foo': 'bar'} # 1 2023-01-01 2 567.8 None ```Extensible Connectors Interface
Add custom connector types with the@make_connector decorator.
```python
from meerschaum.connectors import make_connector, Connector
required = ['requests']
@make_connector
class NWSConnector(Connector):
REQUIRED_ATTRIBUTES = ['username', 'password']
def fetch(self, pipe, begin=None, end=None, **kw):
params = {}
begin = begin or pipe.get_sync_time()
if begin:
params['start'] = begin.isoformat()
if end:
params['end'] = end.isoformat()
stations = pipe.parameters.get('nws', {}).get('stations', [])
for station in stations:
url = f"https://api.weather.gov/stations/{station}/observations"
response = self.session.get(url, params=params)
yield [
feature['properties']
for feature in response.json()['features']
]
@property
def session(self):
_sesh = self.__dict__.get('_session', None)
if _sesh is not None:
return _sesh
import requests
self._session = requests.Session()
self._session.auth = (self.username, self.password)
return self._session
```