如何用Informatica PowerCenter/Unix/Python处理文本中的JSON字符串？

一、Python 处理方案 #

1. 修正JSON语法错误 #

针对引号不匹配、末尾多余逗号这类常见JSON语法错误，可借助jsonrepair库自动修复：

import json
from jsonrepair import repair_json

# 读取存在语法错误的JSON文件
with open('input.json', 'r', encoding='utf-8') as f:
    invalid_json = f.read()

# 修复并解析JSON
fixed_json = repair_json(invalid_json)
parsed_data = json.loads(fixed_json)

2. 去除特殊字符 #

根据需求定义要过滤的特殊字符，用正则表达式批量清理：

import re

cleaned_data = {}
for key, value in parsed_data.items():
    if isinstance(value, str):
        # 示例：去除非打印字符，保留字母数字、空格、逗号、句号
        cleaned_value = re.sub(r'[\x00-\x1F\x7F-\x9F]', '', value)
        cleaned_value = re.sub(r'[^\w\s.,]', '', cleaned_value)
        cleaned_data[key] = cleaned_value
    else:
        cleaned_data[key] = value

3. 转换为指定格式并写入文件 #

格式1：格式化JSON

with open('output_formatted.json', 'w', encoding='utf-8') as f:
    json.dump(cleaned_data, f, indent=4, ensure_ascii=False)

格式2：CSV格式（适用于扁平化JSON结构）

import csv

# 提取字段作为CSV表头
headers = cleaned_data.keys()
with open('output.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.DictWriter(f, fieldnames=headers)
    writer.writeheader()
    writer.writerow(cleaned_data)

二、Informatica PowerCenter 处理方案 #

1. 修正JSON语法错误 #

Informatica无内置JSON修复工具，可先通过Python预脚本修复，或在表达式转换中用正则修复常见问题：

• 去除末尾多余逗号：REG_REPLACE(in_json, ',\s*([}\]])', '\1')
• 补全未闭合的引号：REG_REPLACE(in_json, '([^\\])"([^"]*[^\\])"([^"]*)$', '\1"\2\3"')

2. 去除特殊字符 #

在表达式转换中使用REG_REPLACE函数过滤：

REG_REPLACE(json_field, '[\x00-\x1F\x7F-\x9F]', '') -- 移除非打印字符
REG_REPLACE(json_field, '[^\w\s.,]', '') -- 仅保留指定字符，删除其他特殊符号

3. 解析并转换输出格式 #

1. 使用JSON Parser转换将清理后的JSON字段解析为结构化数据。
2. 添加Router转换，将数据路由到两个输出分支：
   • 分支1：用Expression转换拼接成格式化JSON字符串，写入文件目标。
   • 分支2：用Expression转换扁平化数据结构，写入CSV文件目标。

内容的提问来源于stack exchange，提问作者Sam