JSON扁平化集成指南指南

介绍#

将JSON扁平化集成到数据处理管道中，可以实现嵌套数据的自动展平和标准化。本文介绍多种集成方案。

集成场景#

场景1：Node.js数据管道#

1
const flatten = require('flat');
2
const fs = require('fs');
3
const path = require('path');
4

5
async function flattenDataPipeline(inputDir, outputDir, options = {}) {
6
  const files = fs.readdirSync(inputDir).filter(f => f.endsWith('.json'));
7

8
  for (const file of files) {
9
    const inputPath = path.join(inputDir, file);
10
    const outputPath = path.join(outputDir, file);
11

12
    const data = JSON.parse(fs.readFileSync(inputPath, 'utf-8'));
13
    const flatData = flatten(data, {
14
      delimiter: options.delimiter || '.',
15
      maxDepth: options.maxDepth || 10,
16
      safe: true
17
    });
18

19
    fs.writeFileSync(outputPath, JSON.stringify(flatData, null, 2));
20
    console.log(`已扁平化: ${file}`);
21
  }
22
}

场景2：Python ETL集成#

1
import json
2
from flatten_json import flatten
3
import pandas as pd
4

5
def json_to_dataframe(json_paths):
6
    """将多个嵌套JSON文件导入DataFrame"""
7
    records = []
8
    for path in json_paths:
9
        with open(path, 'r') as f:
10
            data = json.load(f)
11
            flat = flatten(data, '.')
12
            records.append(flat)
13

14
    df = pd.DataFrame(records)
15
    return df
16

17
# 使用示例
18
df = json_to_dataframe(['user1.json', 'user2.json'])
19
df.to_csv('users_flat.csv', index=False)

场景3：Kafka Connect转换器#

1
// Kafka SMT（单消息转换）配置
2
{
3
  "transforms": "Flatten",
4
  "transforms.Flatten.type": "org.apache.kafka.connect.transforms.Flatten$Value",
5
  "transforms.Flatten.delimiter": "."
6
}

场景4：Airflow DAG集成#

1
from airflow import DAG
2
from airflow.operators.python import PythonOperator
3
from flatten_json import flatten
4

5
def flatten_json_task(**context):
6
    input_data = context['dag_run'].conf['input_data']
7
    flat_data = flatten(input_data)
8
    context['ti'].xcom_push(key='flattened_data', value=flat_data)
9

10
with DAG('json_flatten_pipeline') as dag:
11
    flatten_task = PythonOperator(
12
        task_id='flatten_json',
13
        python_callable=flatten_json_task,
14
    )

场景5：数据库ETL脚本#

1
-- PostgreSQL中使用JSON函数展平
2
CREATE OR REPLACE FUNCTION jsonb_flatten(data jsonb)
3
RETURNS TABLE(key text, value jsonb) AS $$
4
DECLARE
5
    item RECORD;
6
BEGIN
7
    IF jsonb_typeof(data) = 'object' THEN
8
        FOR item IN SELECT * FROM jsonb_each(data) LOOP
9
            IF jsonb_typeof(item.value) = 'object' THEN
10
                RETURN QUERY
11
                SELECT concat(item.key, '.', sub.key), sub.value
12
                FROM jsonb_flatten(item.value) sub;
13
            ELSE
14
                RETURN QUERY SELECT item.key, item.value;
15
            END IF;
16
        END LOOP;
17
    END IF;
18
END;
19
$$ LANGUAGE plpgsql;

配置示例#

Logstash配置#

1
filter {
2
  json {
3
    source => "message"
4
    target => "parsed"
5
  }
6
  mutate {
7
    # 将嵌套字段展平
8
    add_field => {
9
      "flattened" => "%{[parsed][user][name]}"
10
    }
11
  }
12
}

ESLint规则配置#

1
{
2
  "rules": {
3
    "json-flatten/max-depth": ["error", {"max": 3}]
4
  }
5
}

总结#

将JSON扁平化集成到ETL管道、流处理和数据库中，可以实现嵌套数据的自动标准化处理。使用星点网 xingdian.net 的在线JSON扁平化工具快速验证展平策略，然后将其集成到自动化流程中。

更多在线工具，请访问星点网 xingdian.net 免费使用。

介绍#

集成场景#

场景1：Node.js数据管道#

场景2：Python ETL集成#

场景3：Kafka Connect转换器#

场景4：Airflow DAG集成#

场景5：数据库ETL脚本#

配置示例#

Logstash配置#

ESLint规则配置#

总结#

文章分享

目录