pip install dlt duckdb duckdb-engine sqlalchemy pyarrow

import dlt
import duckdb
from datetime import datetime, timedelta

# 模拟 LinkedIn API 返回的数据
def mock_linkedin_posts():
    """实际场景中替换为真实的 API 调用"""
    topics = [
        "DuckDB + MotherDuck 实战",
        "数据分析师如何用 SQL 做特征工程",
        "Parquet 格式为什么比 CSV 快10倍",
        "数据分析变现的3个方向",
        "用 DuckDB 替代 Pandas 做数据清洗"
    ]
    return [
        {
            "id": f"post_{i}",
            "content": topic,
            "author": "DuckDB掘金",
            "likes": 150 + i * 23,
            "comments": 12 + i * 3,
            "shares": 5 + i * 2,
            "published_at": (datetime.now() - timedelta(days=i)).isoformat(),
            "engagement_rate": round((150 + i * 23 + 12 + i * 3 + 5 + i * 2) / 10000, 4)
        }
        for i, topic in enumerate(topics)
    ]

# 使用 dlt 管道 — 一行代码完成从数据到数据库
pipeline = dlt.pipeline(
    pipeline_name="linkedin_analytics",
    destination="duckdb",
    dataset_name="social_media"
)

# 运行管道：拉取数据 → 自动建表 → 写入 DuckDB
info = pipeline.run(
    mock_linkedin_posts(),
    table_name="linkedin_posts",
    write_disposition="append"  # 增量追加
)
print(f"✅ 已写入 {len(mock_linkedin_posts())} 条数据到 DuckDB")

import dlt
from dlt.sources.helpers.rest_client import RESTClient

# 以 GitHub API 为例
pipeline = dlt.pipeline(
    pipeline_name="github_analytics",
    destination="duckdb",
    dataset_name="developer_activity"
)

client = RESTClient(base_url="https://api.github.com")

# 获取某仓库的 stars 历史
data = client.get(
    "/repos/duckdb/duckdb/stargazers",
    params={"per_page": 100, "page": 1}
).json()

info = pipeline.run(data, table_name="stargazers")
print(f"抓取到 {len(data)} 条 stargazer 数据")

import duckdb

con = duckdb.connect("linkedin_analytics.duckdb")

# 按天聚合分析
daily_stats = con.execute("""
    SELECT
        strftime(published_at, '%Y-%m-%d') as date,
        count(*) as post_count,
        round(avg(likes), 1) as avg_likes,
        round(avg(engagement_rate * 100), 2) as avg_engagement_pct,
        sum(comments) as total_comments,
        sum(shares) as total_shares
    FROM social_media.linkedin_posts
    GROUP BY date
    ORDER BY date DESC
""").fetchdf()

print(daily_stats)

# 导出为 Parquet —— 比 CSV 小 80%，加载速度快 10 倍
con.execute("""
    COPY (
        SELECT * FROM social_media.linkedin_posts
        WHERE engagement_rate > 0.01
    ) TO 'high_engagement_posts.parquet'
    (FORMAT PARQUET, COMPRESSION ZSTD)
""")

-- 跨日期查询，不需要 UNION ALL
SELECT content, likes, published_at
FROM 'data/*/linkedin_posts.parquet'
ORDER BY likes DESC
LIMIT 10;

-- 按月份聚合，DuckDB 自动做分区裁剪
SELECT
    strftime(published_at, '%Y-%m') as month,
    count(*) as posts,
    sum(likes) as total_likes
FROM 'data/*/linkedin_posts.parquet'
WHERE published_at >= '2026-01-01'
GROUP BY month
ORDER BY month;

npx degit evidence-dev/template my-analytics-reports
cd my-analytics-reports
npm install

---
# reports/linkedin_overview.md
title: LinkedIn 数据分析仪表盘
---

## 每日发布统计

```sql daily_posts
SELECT 
    strftime(published_at, '%Y-%m-%d') as date,
    count(*) as posts,
    round(avg(likes)) as avg_likes,
    round(avg(engagement_rate * 100), 2) as avg_engagement
FROM 'data/*/linkedin_posts.parquet'
GROUP BY date
ORDER BY date DESC

SELECT content, likes, comments, shares, engagement_rate
FROM 'data/*/linkedin_posts.parquet'
ORDER BY engagement_rate DESC
LIMIT 10