Meltano has established itself as a robust alternative to proprietary ETL solutions, but many organizations struggle to move beyond basic use cases. This post is a sequel to our previous article about Meltano which explores advanced patterns, optimization techniques, and architectural considerations for deploying Meltano in production environments at scale.<\/p>\n
While the Singer ecosystem provides extensive coverage, production environments frequently require custom extractors for proprietary systems or specialized data sources.<\/p>\n
Custom taps follow the Singer protocol, emitting RECORD, STATE, and SCHEMA messages as JSON-formatted streams. Here’s a minimal implementation pattern:<\/p>\n
# custom_tap.py\r\nimport json\r\nimport sys\r\nfrom datetime import datetime, timedelta\r\nimport requests\r\n\r\nclass CustomTap:\r\n def __init__(self, config):\r\n self.config = config\r\n self.session = requests.Session()\r\n self.session.headers.update({\r\n 'Authorization': f'Bearer {config[\"api_token\"]}'\r\n })\r\n \r\n def discover(self):\r\n \"\"\"Emit schema for available tables\"\"\"\r\n schema = {\r\n \"type\": \"object\",\r\n \"properties\": {\r\n \"id\": {\"type\": \"integer\"},\r\n \"name\": {\"type\": \"string\"},\r\n \"created_at\": {\"type\": \"string\", \"format\": \"date-time\"},\r\n \"metadata\": {\"type\": \"object\"}\r\n },\r\n \"key_properties\": [\"id\"]\r\n }\r\n catalog = {\r\n \"streams\": [{\r\n \"tap_stream_id\": \"orders\",\r\n \"key_properties\": [\"id\"],\r\n \"schema\": schema,\r\n \"replication_key\": \"created_at\",\r\n \"replication_method\": \"INCREMENTAL\"\r\n }]\r\n }\r\n print(json.dumps(catalog))\r\n \r\n def sync(self, state=None):\r\n \"\"\"Extract data using incremental replication\"\"\"\r\n last_sync = state.get(\"bookmarks\", {}).get(\"orders\", {}).get(\"last_sync\")\r\n start_date = last_sync or self.config[\"start_date\"]\r\n \r\n page = 1\r\n while True:\r\n response = self.session.get(\r\n f\"{self.config['base_url']}\/orders\",\r\n params={\"start_date\": start_date, \"page\": page}\r\n )\r\n response.raise_for_status()\r\n data = response.json()\r\n \r\n if not data[\"records\"]:\r\n break\r\n \r\n for record in data[\"records\"]:\r\n print(json.dumps({\r\n \"type\": \"RECORD\",\r\n \"stream\": \"orders\",\r\n \"record\": record,\r\n \"time_extracted\": datetime.utcnow().isoformat()\r\n }))\r\n state[\"bookmarks\"][\"orders\"][\"last_sync\"] = record[\"created_at\"]\r\n \r\n page += 1\r\n \r\n print(json.dumps({\r\n \"type\": \"STATE\",\r\n \"value\": state\r\n }))\r\n\r\nif __name__ == \"__main__\":\r\n import argparse\r\n parser = argparse.ArgumentParser()\r\n parser.add_argument(\"-c\", \"--config\", required=True)\r\n parser.add_argument(\"-d\", \"--discover\", action=\"store_true\")\r\n parser.add_argument(\"-s\", \"--state\", default=\"{}\")\r\n args = parser.parse_args()\r\n \r\n with open(args.config) as f:\r\n config = json.load(f)\r\n \r\n tap = CustomTap(config)\r\n if args.discover:\r\n tap.discover()\r\n else:\r\n state = json.loads(args.state)\r\n tap.sync(state)\r\n<\/code><\/pre>\nThis pattern enables incremental replication with state management, critical for handling large datasets efficiently.<\/p>\n
Advanced Meltano Configuration and Plugin Development<\/h2>\nPlugin Metadata and Custom Properties<\/h3>\n
Meltano’s plugin system extends beyond standard connectors. Custom plugins with metadata-driven behavior provide flexibility:<\/p>\n
plugins:\r\n extractors:\r\n - name: tap-custom-api\r\n variant: custom\r\n pip_url: git+https:\/\/github.com\/org\/tap-custom-api.git@main\r\n executable: tap-custom-api\r\n config:\r\n base_url: https:\/\/api.example.com\r\n request_timeout: 30\r\n max_retries: 3\r\n backoff_strategy: exponential\r\n settings:\r\n - name: api_token\r\n kind: password\r\n value: ${CUSTOM_API_TOKEN}\r\n - name: start_date\r\n kind: date_iso8601\r\n value: 2023-01-01T00:00:00Z\r\n - name: batch_size\r\n kind: integer\r\n value: 1000\r\n \r\n transformers:\r\n - name: dbt-custom\r\n variant: dbt-postgres\r\n pip_url: dbt-core~=1.6.0 dbt-postgres~=1.6.0\r\n commands:\r\n debug:\r\n args: debug\r\n run:\r\n args: run --select state:modified+\r\n container: false\r\n test:\r\n args: test --fail-fast\r\n<\/code><\/pre>\nCustom Mappers for Data Normalization<\/h3>\n
Mappers transform data stream structure before loading. Create custom mappers for domain-specific transformations:<\/p>\n
# custom_mapper.py\r\nimport json\r\nimport sys\r\nfrom typing import Dict, Any\r\n\r\nclass CustomMapper:\r\n \"\"\"Flatten nested objects and normalize field names\"\"\"\r\n \r\n def __init__(self):\r\n self.field_mapping = {\r\n \"user_info.email\": \"user_email\",\r\n \"user_info.created_at\": \"user_created_at\",\r\n \"metadata.tags\": \"tags\"\r\n }\r\n \r\n def transform_record(self, record: Dict[str, Any]) -> Dict[str, Any]:\r\n \"\"\"Flatten and normalize record structure\"\"\"\r\n result = {}\r\n \r\n # Handle nested paths\r\n for target_field, source_path in self.field_mapping.items():\r\n value = self._get_nested(record, source_path)\r\n if value is not None:\r\n result[target_field] = value\r\n \r\n # Copy remaining top-level fields\r\n for key, value in record.items():\r\n if key not in self.field_mapping.values():\r\n result[key] = value\r\n \r\n return result\r\n \r\n @staticmethod\r\n def _get_nested(obj: Dict, path: str) -> Any:\r\n \"\"\"Access nested dictionary values using dot notation\"\"\"\r\n for key in path.split('.'):\r\n if isinstance(obj, dict):\r\n obj = obj.get(key)\r\n else:\r\n return None\r\n return obj\r\n\r\nmapper = CustomMapper()\r\n\r\nfor line in sys.stdin:\r\n msg = json.loads(line)\r\n \r\n if msg[\"type\"] == \"RECORD\":\r\n msg[\"record\"] = mapper.transform_record(msg[\"record\"])\r\n \r\n print(json.dumps(msg))\r\n<\/code><\/pre>\nState Management and Incremental Replication at Scale<\/h2>\n
Proper state management is critical for scalable pipelines. Beyond simple bookmarks, advanced strategies handle complex scenarios:<\/p>\n
Distributed State Across Multiple Streams<\/h3>\n# advanced_state_management.py\r\nimport json\r\nfrom datetime import datetime, timedelta\r\nfrom concurrent.futures import ThreadPoolExecutor, as_completed\r\n\r\nclass DistributedStateManager:\r\n def __init__(self, state_backend=\"postgres\"):\r\n self.state_backend = state_backend\r\n self.batch_states = {}\r\n \r\n def create_partition_state(self, stream_name: str, partition_key: str, \r\n value: Any) -> Dict:\r\n \"\"\"Create granular state for partitioned data\"\"\"\r\n return {\r\n \"stream\": stream_name,\r\n \"partition\": partition_key,\r\n \"value\": value,\r\n \"updated_at\": datetime.utcnow().isoformat(),\r\n \"version\": 1\r\n }\r\n \r\n def batch_process_with_checkpoints(self, extractor, batch_size: int = 10000):\r\n \"\"\"Process data in batches with intermediate state commits\"\"\"\r\n batch = []\r\n batch_number = 0\r\n \r\n for record in extractor.extract():\r\n batch.append(record)\r\n \r\n if len(batch) >= batch_size:\r\n yield {\r\n \"type\": \"BATCH\",\r\n \"batch_number\": batch_number,\r\n \"records\": batch,\r\n \"checkpoint\": {\r\n \"processed\": batch_size * (batch_number + 1),\r\n \"timestamp\": datetime.utcnow().isoformat()\r\n }\r\n }\r\n batch = []\r\n batch_number += 1\r\n \r\n # Final batch\r\n if batch:\r\n yield {\r\n \"type\": \"BATCH\",\r\n \"batch_number\": batch_number,\r\n \"records\": batch,\r\n \"checkpoint\": {\r\n \"processed\": batch_size * batch_number + len(batch),\r\n \"timestamp\": datetime.utcnow().isoformat()\r\n }\r\n }\r\n<\/code><\/pre>\nPerformance Optimization Techniques<\/h2>\nParallel Tap Execution with Process Pooling<\/h3>\n
For independent data sources, parallel extraction significantly reduces runtime:<\/p>\n
# meltano.yml\r\nrun_steps:\r\n parallel_extract:\r\n commands:\r\n - meltano run tap-salesforce target-warehouse\r\n - meltano run tap-stripe target-warehouse\r\n - meltano run tap-hubspot target-warehouse\r\n\r\npipelines:\r\n - id: multi-source-parallel\r\n steps:\r\n - run_step: parallel_extract\r\n<\/code><\/pre>\nCombined with run_dir<\/code> configurations for isolation:<\/p>\nmeltano run --jobs 3 \\\r\n tap-salesforce target-warehouse \\\r\n tap-stripe target-warehouse \\\r\n tap-hubspot target-warehouse\r\n<\/code><\/pre>\nQuery Optimization in Taps<\/h3>\n
Implement efficient data extraction patterns:<\/p>\n
def extract_with_partitioning(self, start_date, end_date, partition_days=7):\r\n \"\"\"Extract large datasets using time-based partitioning\"\"\"\r\n current = start_date\r\n \r\n while current < end_date:\r\n partition_end = min(current + timedelta(days=partition_days), end_date)\r\n \r\n records = self.query_api(\r\n f\"SELECT * FROM orders WHERE created_at >= '{current}' AND created_at < '{partition_end}'\",\r\n order_by=\"created_at ASC\",\r\n page_size=5000\r\n )\r\n \r\n for record in records:\r\n yield record\r\n \r\n current = partition_end\r\n<\/code><\/pre>\nAdvanced dbt Integration and Testing<\/h2>\nSemantic Versioning with dbt<\/h3>\n
Implement versioned models for backward compatibility:<\/p>\n
-- models\/marts\/fact_orders_v1.sql\r\n{{ config(\r\n version=1,\r\n materialized='table',\r\n on_schema_change='fail'\r\n) }}\r\n\r\nSELECT \r\n order_id,\r\n customer_id,\r\n order_total,\r\n created_at\r\nFROM {{ ref('stg_orders') }}\r\n\r\n-- models\/marts\/fact_orders_v2.sql\r\n{{ config(\r\n version=2,\r\n materialized='table',\r\n on_schema_change='fail'\r\n) }}\r\n\r\nSELECT \r\n order_id,\r\n customer_id,\r\n order_total,\r\n tax_amount,\r\n net_total,\r\n created_at,\r\n updated_at\r\nFROM {{ ref('stg_orders') }}\r\nWHERE deleted_at IS NULL\r\n<\/code><\/pre>\nContract Testing in dbt<\/h3>\n
Define and enforce data contracts:<\/p>\n
# models\/marts\/fact_orders.yml\r\nmodels:\r\n - name: fact_orders\r\n config:\r\n contract:\r\n enforced: true\r\n \r\n columns:\r\n - name: order_id\r\n data_tests:\r\n - not_null\r\n - unique\r\n description: Primary key\r\n data_type: bigint\r\n \r\n - name: customer_id\r\n data_tests:\r\n - not_null\r\n - relationships:\r\n to: ref('dim_customers')\r\n field: customer_id\r\n data_type: bigint\r\n \r\n - name: order_total\r\n data_tests:\r\n - dbt_utils.expression_is_true:\r\n expression: \"order_total >= 0\"\r\n data_type: numeric\r\n<\/code><\/pre>\nProduction Deployment Patterns<\/h2>\nEnvironment-Specific Configuration<\/h3>\n# meltano.yml with environment overrides\r\nenv:\r\n dev:\r\n MELTANO_ENVIRONMENT: dev\r\n PG_HOST: localhost\r\n PG_PORT: 5432\r\n PG_DATABASE: meltano_dev\r\n \r\n staging:\r\n MELTANO_ENVIRONMENT: staging\r\n PG_HOST: postgres-staging.internal\r\n PG_PORT: 5432\r\n PG_DATABASE: meltano_staging\r\n MELTANO_LOG_LEVEL: debug\r\n \r\n prod:\r\n MELTANO_ENVIRONMENT: prod\r\n PG_HOST: postgres-prod.internal\r\n PG_PORT: 5432\r\n PG_DATABASE: meltano_prod\r\n MELTANO_LOG_LEVEL: info\r\n<\/code><\/pre>\nHealth Checks and Monitoring<\/h3>\n# monitoring.py\r\nimport logging\r\nimport json\r\nfrom datetime import datetime, timedelta\r\nfrom dataclasses import dataclass, asdict\r\n\r\n@dataclass\r\nclass PipelineHealthMetrics:\r\n pipeline_name: str\r\n status: str # success, warning, failure\r\n records_extracted: int\r\n records_loaded: int\r\n duration_seconds: float\r\n last_run: str\r\n next_run: str\r\n error_message: str = None\r\n\r\nclass PipelineMonitor:\r\n def __init__(self, metrics_backend):\r\n self.backend = metrics_backend\r\n self.logger = logging.getLogger(__name__)\r\n \r\n def record_pipeline_execution(self, metrics: PipelineHealthMetrics):\r\n \"\"\"Record pipeline metrics for monitoring\"\"\"\r\n self.backend.insert(\"pipeline_metrics\", asdict(metrics))\r\n \r\n # Alert if extraction failed\r\n if metrics.status == \"failure\":\r\n self.alert(f\"Pipeline {metrics.pipeline_name} failed: {metrics.error_message}\")\r\n \r\n # Warn if significantly slower than average\r\n avg_duration = self.backend.get_average_duration(metrics.pipeline_name)\r\n if metrics.duration_seconds > avg_duration * 1.5:\r\n self.logger.warning(f\"Pipeline {metrics.pipeline_name} running slow\")\r\n \r\n def validate_freshness(self, pipeline_name: str, max_age_hours: int):\r\n \"\"\"Verify data is within SLA\"\"\"\r\n last_run = self.backend.get_last_run(pipeline_name)\r\n age = datetime.utcnow() - last_run\r\n \r\n if age > timedelta(hours=max_age_hours):\r\n self.alert(f\"Pipeline {pipeline_name} data is stale\")\r\n return False\r\n return True\r\n<\/code><\/pre>\nOrchestration with Airflow<\/h3>\n# dags\/meltano_pipeline.py\r\nfrom airflow import DAG\r\nfrom airflow.operators.bash import BashOperator\r\nfrom airflow.sensors.external_task import ExternalTaskSensor\r\nfrom datetime import datetime, timedelta\r\n\r\ndefault_args = {\r\n 'owner': 'data-engineering',\r\n 'retries': 2,\r\n 'retry_delay': timedelta(minutes=5),\r\n}\r\n\r\ndag = DAG(\r\n 'meltano_multi_source',\r\n default_args=default_args,\r\n schedule_interval='0 2 * * *',\r\n max_active_runs=1,\r\n tags=['meltano', 'production']\r\n)\r\n\r\nextract_salesforce = BashOperator(\r\n task_id='extract_salesforce',\r\n bash_command='meltano run tap-salesforce target-warehouse',\r\n env={'MELTANO_ENVIRONMENT': 'prod'},\r\n dag=dag\r\n)\r\n\r\nextract_stripe = BashOperator(\r\n task_id='extract_stripe',\r\n bash_command='meltano run tap-stripe target-warehouse',\r\n env={'MELTANO_ENVIRONMENT': 'prod'},\r\n dag=dag\r\n)\r\n\r\ntransform = BashOperator(\r\n task_id='transform_dbt',\r\n bash_command='meltano invoke dbt:run --models staging.* marts.*',\r\n env={'MELTANO_ENVIRONMENT': 'prod'},\r\n dag=dag\r\n)\r\n\r\ntest = BashOperator(\r\n task_id='test_dbt',\r\n bash_command='meltano invoke dbt:test --fail-fast',\r\n env={'MELTANO_ENVIRONMENT': 'prod'},\r\n dag=dag\r\n)\r\n\r\n[extract_salesforce, extract_stripe] >> transform >> test\r\n<\/code><\/pre>\nAdvanced Troubleshooting and Debugging<\/h2>\nDeep Logging and Trace Analysis<\/h3>\n# debug_runner.py\r\nimport logging\r\nimport sys\r\nimport json\r\nfrom collections import defaultdict\r\n\r\nclass PipelineDebugger:\r\n def __init__(self, log_level=logging.DEBUG):\r\n self.logger = logging.getLogger('meltano')\r\n self.logger.setLevel(log_level)\r\n self.stats = defaultdict(int)\r\n \r\n def analyze_tap_output(self, stdin):\r\n \"\"\"Analyze Singer messages for anomalies\"\"\"\r\n for line in stdin:\r\n msg = json.loads(line)\r\n msg_type = msg.get('type')\r\n \r\n self.stats[msg_type] += 1\r\n \r\n if msg_type == 'RECORD':\r\n stream = msg.get('stream')\r\n self.logger.debug(f\"Record from {stream}: {len(json.dumps(msg['record']))} bytes\")\r\n \r\n elif msg_type == 'STATE':\r\n self.logger.debug(f\"State checkpoint: {msg['value']}\")\r\n \r\n elif msg_type == 'SCHEMA':\r\n properties = msg['schema'].get('properties', {})\r\n self.logger.debug(f\"Schema fields: {list(properties.keys())}\")\r\n \r\n yield msg\r\n \r\n def print_summary(self):\r\n \"\"\"Report pipeline statistics\"\"\"\r\n print(\"\\n=== Pipeline Summary ===\")\r\n for msg_type, count in sorted(self.stats.items()):\r\n print(f\"{msg_type}: {count}\")\r\n<\/code><\/pre>\nConclusion<\/h2>\n
Production Meltano deployments require careful attention to custom development, state management, performance optimization, and observability. By combining Meltano’s flexibility with advanced patterns\u2014custom taps, parallel execution, semantic versioning in dbt, and comprehensive monitoring\u2014organizations can build scalable, maintainable data platforms that rival proprietary solutions while maintaining the agility of open-source tools.<\/p>\n
The key is starting with solid fundamentals, measuring performance, and iterating on your architecture as requirements evolve.<\/p>\n","protected":false},"excerpt":{"rendered":"
Meltano has established itself as a robust alternative to proprietary ETL solutions, but many organizations struggle to move beyond basic use cases. This post is a sequel to our previous…<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[43,45],"tags":[47,39,46],"class_list":["post-63","post","type-post","status-publish","format-standard","hentry","category-etl","category-meltano","tag-data-engineering","tag-etl","tag-meltano"],"_links":{"self":[{"href":"https:\/\/biziq.info\/index.php\/wp-json\/wp\/v2\/posts\/63","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/biziq.info\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/biziq.info\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/biziq.info\/index.php\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/biziq.info\/index.php\/wp-json\/wp\/v2\/comments?post=63"}],"version-history":[{"count":1,"href":"https:\/\/biziq.info\/index.php\/wp-json\/wp\/v2\/posts\/63\/revisions"}],"predecessor-version":[{"id":64,"href":"https:\/\/biziq.info\/index.php\/wp-json\/wp\/v2\/posts\/63\/revisions\/64"}],"wp:attachment":[{"href":"https:\/\/biziq.info\/index.php\/wp-json\/wp\/v2\/media?parent=63"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/biziq.info\/index.php\/wp-json\/wp\/v2\/categories?post=63"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/biziq.info\/index.php\/wp-json\/wp\/v2\/tags?post=63"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}