Menu
The review-first prompt is selected by default. You can switch to a direct command or download a local copy.
Read SKILL.md and any companion files shown by SkillsMP before deciding whether to install.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
npx skills add https://github.com/ffsshhttiikk/opencode-agents-skills --skill etl-pipelinesThe command stays on one line. Scroll horizontally to inspect it before copying.
Prefer a local copy? Download the files currently available to SkillsMP.
Based on SOC occupation classification
Showing SKILL.md
| name | etl-pipelines |
| description | ETL pipeline design and implementation |
| license | MIT |
| compatibility | opencode |
| metadata | {"audience":"data-engineers","category":"data-science"} |
Use me when:
# Apache Airflow DAG
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
default_args = {
"owner": "data_team",
"depends_on_past": False,
"retries": 3,
"retry_delay": timedelta(minutes=5)
}
with DAG(
"etl_pipeline",
start_date=datetime(2024, 1, 1),
schedule_interval="@daily",
default_args=default_args
) as dag:
extract = PythonOperator(
task_id="extract",
python_callable=extract_data
)
transform = PythonOperator(
task_id="transform",
python_callable=transform_data,
dependencies=[extract]
)
load = PythonOperator(
task_id="load",
python_callable=load_data,
dependencies=[transform]
)
from pyspark.sql import SparkSession
from pyspark.sql.functions import window, col
spark = SparkSession.builder \
.appName("StreamingETL") \
.config("spark.sql.streaming.checkpointLocation", "/checkpoints") \
.getOrCreate()
# Read streaming data
stream = spark.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "broker:9092") \
.option("subscribe", "events") \
.load()
# Transform streaming
enriched = stream \
.select(
col("value").cast("string").alias("json")
) \
.withColumn("data", F.from_json("json", schema)) \
.select("data.*")
# Write streaming
query = enriched \
.writeStream \
.format("delta") \
.option("checkpointLocation", "/checkpoints") \
.trigger(processingTime="1 minute") \
.start("/data/lakehouse/tables/events")