ingesting-data

Name: ingesting-data
Author: ancoleman

48installs59forks

How to install ingesting-data?

ancoleman/ai-design-components/ingesting-data

$npx -y skills add ancoleman/ai-design-components --skill ingesting-data

Installs into the current project.

Summary

Data ingestion patterns for loading data from cloud storage, APIs, files, and streaming sources into databases. Use when importing CSV/JSON/Parquet files, pulling from S3/GCS buckets, consuming API feeds, or building ETL pipelines.

Files · 1

View on GitHub

SKILL.md

1	# Data Ingestion Patterns
2
3	This skill provides patterns for getting data INTO systems from external sources.
4
5	## When to Use This Skill
6
7	- Importing CSV, JSON, Parquet, or Excel files
8	- Loading data from S3, GCS, or Azure Blob storage
9	- Consuming REST/GraphQL API feeds
10	- Building ETL/ELT pipelines
11	- Database migration and CDC (Change Data Capture)
12	- Streaming data ingestion from Kafka/Kinesis
13
14	## Ingestion Pattern Decision Tree
15
16	```
17	What is your data source?
18	├── Cloud Storage (S3, GCS, Azure) → See cloud-storage.md
19	├── Files (CSV, JSON, Parquet) → See file-formats.md
20	├── REST/GraphQL APIs → See api-feeds.md
21	├── Streaming (Kafka, Kinesis) → See streaming-sources.md
22	├── Legacy Database → See database-migration.md
23	└── Need full ETL framework → See etl-tools.md
24	```
25
26	## Quick Start by Language
27
28	### Python (Recommended for ETL)
29
30	dlt (data load tool) - Modern Python ETL:
31	```python
32	import dlt
33
34	# Define a source
35	@dlt.source
36	def github_source(repo: str):
37	@dlt.resource(write_disposition="merge", primary_key="id")
38	def issues():
39	response = requests.get(f"https://api.github.com/repos/{repo}/issues")
40	yield response.json()
41	return issues
42
43	# Load to destination
44	pipeline = dlt.pipeline(
45	pipeline_name="github_issues",
46	destination="postgres", # or duckdb, bigquery, snowflake
47	dataset_name="github_data"
48	)
49
50	load_info = pipeline.run(github_source("owner/repo"))
51	print(load_info)
52	```
53
54	Polars for file processing (faster than pandas):
55	```python
56	import polars as pl
57
58	# Read CSV with schema inference
59	df = pl.read_csv("data.csv")
60
61	# Read Parquet (columnar, efficient)
62	df = pl.read_parquet("s3://bucket/data.parquet")
63
64	# Read JSON lines
65	df = pl.read_ndjson("events.jsonl")
66
67	# Write to database
68	df.write_database(
69	table_name="events",
70	connection="postgresql://user:pass@localhost/db",
71	if_table_exists="append"
72	)
73	```
74
75	### TypeScript/Node.js
76
77	S3 ingestion:
78	```typescript
79	import { S3Client, GetObjectCommand } from "@aws-sdk/client-s3";
80	import { parse } from "csv-parse/sync";
81
82	const s3 = new S3Client({ region: "us-east-1" });
83
84	async function ingestFromS3(bucket: string, key: string) {
85	const response = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: key }));
86	const body = await response.Body?.transformToString();
87
88	// Parse CSV
89	const records = parse(body, { columns: true, skip_empty_lines: true });
90
91	// Insert to database
92	await db.insert(eventsTable).values(records);
93	}
94	```
95
96	API feed polling:
97	```typescript
98	import { Hono } from "hono";
99
100	// Webhook receiver for real-time ingestion
101	const app = new Hono();
102
103	app.post("/webhooks/stripe", async (c) => {
104	const event = await c.req.json();
105
106	// Validate webhook signature
107	const signature = c.req.header("stripe-signature");
108	// ... validation logic
109
110	// Ingest event
111	await db.insert(stripeEventsTable).values({
112	eventId: event.id,
113	type: event.type,
114	data: event.data,
115	receivedAt: new Date()
116	});
117
118	return c.json({ received: true });
119	});
120	```
121
122	### Rust
123
124	High-performance file ingestion:
125	```rust
126	use polars::prelude::*;
127	use aws_sdk_s3::Client;
128
129	async fn ingest_parquet(client: &Client, bucket: &str, key: &str) -> Result<DataFrame> {
130	// Download from S3
131	let resp = client.get_object()
132	.bucket(bucket)
133	.key(key)
134	.send()
135	.await?;
136
137	let bytes = resp.body.collect().await?.into_bytes();
138
139	// Parse with Polars
140	let df = ParquetReader::new(Cursor::new(bytes))
141	.finish()?;
142
143	Ok(df)
144	}
145	```
146
147	### Go
148
149	Concurrent file processing:
150	```go
151	package main
152
153	import (
154	"context"
155	"encoding/csv"
156	"github.com/aws/aws-sdk-go-v2/service/s3"
157	)
158
159	func ingestCSV(ctx context.Context, client *s3.Client, bucket, key string) error {
160	resp, err := client.GetObject(ctx, &s3.GetObjectInput{
161	Bucket: &bucket,
162	Key: &key,
163	})
164	if err != nil {
165	return err
166	}
167	defer resp.Body.Close()
168
169	reader := csv.NewReader(resp.Body)
170	records, err := reader.ReadAll()
171	if err != nil {
172	return err
173	}
174
175	// Batch insert to database
176	return batchInsert(ctx, records)
177	}
178	```
179
180	## Ingestion Patterns
181
182	### 1. Batch Ingestion (Files/Storage)
183
184	For periodic bulk loads:
185
186	```
187	Source → Extract → Transform → Load → Validate
188	↓ ↓ ↓ ↓ ↓
189	S3 Download Clean/Map Insert Count check
190	```
191
192	Key considerations:
193	- Use chunked reading for large files (>100MB)
194	- Implement idempotency with checksums
195	- Track file processing state
196	- Handle partial failures
197
198	### 2. Streaming Ingestion (Real-time)
199
200	For continuous data flow:
201
202	```
203	Source → Buffer → Process → Load → Ack
204	↓ ↓ ↓ ↓ ↓
205	Kafka In-memory Transform DB Commit offset
206	```

People who liked this also viewed

6 picks

Type

ingesting-data

48installs59forks

View on GitHub

How to install ingesting-data?

ancoleman/ai-design-components/ingesting-data

$npx -y skills add ancoleman/ai-design-components --skill ingesting-data

Installs into the current project.

Summary

Files · 1

View on GitHub

SKILL.md

1	# Data Ingestion Patterns
2
3	This skill provides patterns for getting data INTO systems from external sources.
4
5	## When to Use This Skill
6
7	- Importing CSV, JSON, Parquet, or Excel files
8	- Loading data from S3, GCS, or Azure Blob storage
9	- Consuming REST/GraphQL API feeds
10	- Building ETL/ELT pipelines
11	- Database migration and CDC (Change Data Capture)
12	- Streaming data ingestion from Kafka/Kinesis
13
14	## Ingestion Pattern Decision Tree
15
16	```
17	What is your data source?
18	├── Cloud Storage (S3, GCS, Azure) → See cloud-storage.md
19	├── Files (CSV, JSON, Parquet) → See file-formats.md
20	├── REST/GraphQL APIs → See api-feeds.md
21	├── Streaming (Kafka, Kinesis) → See streaming-sources.md
22	├── Legacy Database → See database-migration.md
23	└── Need full ETL framework → See etl-tools.md
24	```
25
26	## Quick Start by Language
27
28	### Python (Recommended for ETL)
29
30	dlt (data load tool) - Modern Python ETL:
31	```python
32	import dlt
33
34	# Define a source
35	@dlt.source
36	def github_source(repo: str):
37	@dlt.resource(write_disposition="merge", primary_key="id")
38	def issues():
39	response = requests.get(f"https://api.github.com/repos/{repo}/issues")
40	yield response.json()
41	return issues
42
43	# Load to destination
44	pipeline = dlt.pipeline(
45	pipeline_name="github_issues",
46	destination="postgres", # or duckdb, bigquery, snowflake
47	dataset_name="github_data"
48	)
49
50	load_info = pipeline.run(github_source("owner/repo"))
51	print(load_info)
52	```
53
54	Polars for file processing (faster than pandas):
55	```python
56	import polars as pl
57
58	# Read CSV with schema inference
59	df = pl.read_csv("data.csv")
60
61	# Read Parquet (columnar, efficient)
62	df = pl.read_parquet("s3://bucket/data.parquet")
63
64	# Read JSON lines
65	df = pl.read_ndjson("events.jsonl")
66
67	# Write to database
68	df.write_database(
69	table_name="events",
70	connection="postgresql://user:pass@localhost/db",
71	if_table_exists="append"
72	)
73	```
74
75	### TypeScript/Node.js
76
77	S3 ingestion:
78	```typescript
79	import { S3Client, GetObjectCommand } from "@aws-sdk/client-s3";
80	import { parse } from "csv-parse/sync";
81
82	const s3 = new S3Client({ region: "us-east-1" });
83
84	async function ingestFromS3(bucket: string, key: string) {
85	const response = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: key }));
86	const body = await response.Body?.transformToString();
87
88	// Parse CSV
89	const records = parse(body, { columns: true, skip_empty_lines: true });
90
91	// Insert to database
92	await db.insert(eventsTable).values(records);
93	}
94	```
95
96	API feed polling:
97	```typescript
98	import { Hono } from "hono";
99
100	// Webhook receiver for real-time ingestion
101	const app = new Hono();
102
103	app.post("/webhooks/stripe", async (c) => {
104	const event = await c.req.json();
105
106	// Validate webhook signature
107	const signature = c.req.header("stripe-signature");
108	// ... validation logic
109
110	// Ingest event
111	await db.insert(stripeEventsTable).values({
112	eventId: event.id,
113	type: event.type,
114	data: event.data,
115	receivedAt: new Date()
116	});
117
118	return c.json({ received: true });
119	});
120	```
121
122	### Rust
123
124	High-performance file ingestion:
125	```rust
126	use polars::prelude::*;
127	use aws_sdk_s3::Client;
128
129	async fn ingest_parquet(client: &Client, bucket: &str, key: &str) -> Result<DataFrame> {
130	// Download from S3
131	let resp = client.get_object()
132	.bucket(bucket)
133	.key(key)
134	.send()
135	.await?;
136
137	let bytes = resp.body.collect().await?.into_bytes();
138
139	// Parse with Polars
140	let df = ParquetReader::new(Cursor::new(bytes))
141	.finish()?;
142
143	Ok(df)
144	}
145	```
146
147	### Go
148
149	Concurrent file processing:
150	```go
151	package main
152
153	import (
154	"context"
155	"encoding/csv"
156	"github.com/aws/aws-sdk-go-v2/service/s3"
157	)
158
159	func ingestCSV(ctx context.Context, client *s3.Client, bucket, key string) error {
160	resp, err := client.GetObject(ctx, &s3.GetObjectInput{
161	Bucket: &bucket,
162	Key: &key,
163	})
164	if err != nil {
165	return err
166	}
167	defer resp.Body.Close()
168
169	reader := csv.NewReader(resp.Body)
170	records, err := reader.ReadAll()
171	if err != nil {
172	return err
173	}
174
175	// Batch insert to database
176	return batchInsert(ctx, records)
177	}
178	```
179
180	## Ingestion Patterns
181
182	### 1. Batch Ingestion (Files/Storage)
183
184	For periodic bulk loads:
185
186	```
187	Source → Extract → Transform → Load → Validate
188	↓ ↓ ↓ ↓ ↓
189	S3 Download Clean/Map Insert Count check
190	```
191
192	Key considerations:
193	- Use chunked reading for large files (>100MB)
194	- Implement idempotency with checksums
195	- Track file processing state
196	- Handle partial failures
197
198	### 2. Streaming Ingestion (Real-time)
199
200	For continuous data flow:
201
202	```
203	Source → Buffer → Process → Load → Ack
204	↓ ↓ ↓ ↓ ↓
205	Kafka In-memory Transform DB Commit offset
206	```

People who liked this also viewed

6 picks

Type