GoldenFlow

已迁移。 此仓库已迁入 benzsevern/goldenmatch 单体仓库，路径为 packages/python/goldenflow (and packages/typescript/goldenflow)/。此仓库已归档；新开发工作将在单体仓库中进行。

GoldenFlow

数据转换工具包——在数据进入工作流之前，对杂乱数据进行标准化、重塑和规范化。 由 Ben Severn 开发。

可处理文件（CSV、Excel、Parquet）、云存储（S3、GCS）或实时数据库。零配置模式会自动检测需要修正的内容。只需一条命令，即可清理 GoldenCheck 发现的问题，并为 GoldenMatch 准备数据。支持 Python 和 TypeScript，且功能完全一致。

# Python
pip install goldenflow
goldenflow transform data.csv

# TypeScript / Node.js
npm install goldenflow
npx goldenflow-js transform data.csv

---

问题

你的数据总会以一些常见的问题形式到来：

• 电话号码有 15 种不同格式
• 日期混用 MM/DD/YYYY 和 YYYY-MM-DD 格式
• 地址中的缩写不一致
• 不同系统中的列名不匹配（“fname”与“first_name”与“given_name”）
• 值中有前导空格、无效 Unicode 字符和弯引号
• 分类值不一致（“USA”、“US”、“United States”）

每位数据工程师都会编写临时脚本来修复这些问题。每个脚本都略有不同，而且都无法复用。

GoldenFlow 让转换变得可复用、可组合且自动化。

---

快速开始

Python

pip install goldenflow

# Auto-transform (zero-config)
goldenflow transform messy_data.csv

# Try the demo first
goldenflow demo
goldenflow transform demo_data.csv -c demo_config.yaml

# With config
goldenflow learn messy_data.csv -o config.yaml
goldenflow transform messy_data.csv -c config.yaml

# Schema mapping
goldenflow map --source system_a.csv --target system_b.csv

# Full pipeline
goldencheck scan data.csv
goldenflow transform data.csv
goldenmatch dedupe data_transformed.csv

TypeScript / Node.js

npm install goldenflow

# Auto-transform (zero-config)
npx goldenflow-js transform messy_data.csv

# Try the demo first
npx goldenflow-js demo
npx goldenflow-js transform demo_data.csv -c demo_config.yaml

# With config
npx goldenflow-js learn messy_data.csv -o config.yaml
npx goldenflow-js transform messy_data.csv -c config.yaml

# Schema mapping
npx goldenflow-js map -s system_a.csv -t system_b.csv

编程方式（TypeScript）

import { TransformEngine } from "goldenflow";

const result = new TransformEngine().transformDf([
  { name: "  JOHN  ", phone: "(555) 123-4567", email: "[email protected]" },
]);
console.log(result.rows[0]);
// { name: "JOHN", phone: "+15551234567", email: "[email protected]" }

---

零配置模式

goldenflow transform customers.csv
# or just:
goldenflow customers.csv

GoldenFlow 会分析每一列，并自动应用安全的转换：

• 去除空格并规范化 Unicode
• 将电话号码标准化为 E.164 格式
• 统一电子邮件地址的大小写
• 解析日期格式并标准化为 ISO 8601
• 规范化邮政编码（补齐前导零、去除 +4）
• 将弯引号替换为直引号
• 通过模糊匹配自动修正分类值中的拼写错误

输出：一份干净的 CSV，并附带一份清单文件，列出应用的每项转换。

customers.csv           -> customers_transformed.csv
                        -> customers_manifest.json

清单是一份审计记录，说明了哪些内容发生了变化、为何变化，以及影响了哪些行。

---

CLI 命令

GoldenFlow 提供 14 条命令。最常用的命令如下：

# Core transforms
goldenflow transform data.csv                    # Zero-config: auto-detect and fix
goldenflow transform data.csv -c config.yaml     # Apply saved config
goldenflow transform data.csv --domain healthcare # Use a domain pack
goldenflow transform data.csv --strict           # Fail on any transform error
goldenflow transform data.csv --llm              # Enable LLM-enhanced corrections
goldenflow data.csv                              # Shorthand: auto-routes to transform

# Schema & profiling
goldenflow map -s a.csv -t b.csv                 # Auto-map schemas between files
goldenflow profile data.csv                      # Show column profiles
goldenflow learn data.csv -o config.yaml         # Generate config from data patterns
goldenflow validate data.csv                     # Dry-run: show what would change
goldenflow diff before.csv after.csv             # Compare pre/post transform

# Continuous & scheduled
goldenflow watch ./data/                         # Auto-transform new/changed files
goldenflow schedule data.csv --every 1h          # Run on a schedule (5m, 1h, 30s...)
goldenflow stream large_file.csv --chunk-size 50000  # Stream-process in batches

# Discovery & history
goldenflow init data.csv                         # Interactive setup wizard
goldenflow demo                                  # Generate sample data to try
goldenflow history                               # Show recent transform runs
goldenflow history -n 50                         # Last 50 runs

# Integrations
goldenflow interactive data.csv                  # Launch TUI
goldenflow serve                                 # REST API for real-time transforms
goldenflow mcp-serve                             # MCP server for Claude Desktop

默认路由

不带子命令运行 goldenflow <file> 时，会自动路由到 transform：

goldenflow customers.csv       # equivalent to: goldenflow transform customers.csv
goldenflow -                   # read from stdin, write to stdout

---

流式处理

对于大到无法载入内存的文件，请使用 StreamProcessor 或 stream 命令：

goldenflow stream large_file.csv --chunk-size 50000

from goldenflow.streaming import StreamProcessor

processor = StreamProcessor(config=config)

# Process a single record
result = processor.transform_one({"name": "  John  ", "phone": "(555) 123-4567"})

# Process a batch
result = processor.transform_batch(df_batch)

# Stream a large file in chunks
for result in processor.stream_file("large_data.csv", chunk_size=10_000):
    write_to_output(result.df)

print(f"Processed {processor.batches_processed} batches")

---

云连接器

GoldenFlow 可直接读写 S3 和 Google Cloud Storage：

# S3
goldenflow transform s3://my-bucket/raw/customers.csv -o s3://my-bucket/clean/

# GCS
goldenflow transform gs://my-bucket/data/records.csv

from goldenflow.connectors.s3 import read_s3, write_s3
from goldenflow.connectors.gcs import read_gcs, write_gcs

df = read_s3("s3://my-bucket/raw/customers.csv")
df = read_gcs("gs://my-bucket/data/records.csv")

系统会自动识别云路径，无需额外标志。

---

监视模式

文件到达目录时自动转换：

goldenflow watch ./data/
goldenflow watch ./incoming/ -c config.yaml -o ./processed/

GoldenFlow 会轮询目录，并对其中新增或已更改的文件应用转换。

---

调度

按重复计划运行转换：

goldenflow schedule data.csv --every 1h
goldenflow schedule data.csv --every 30m -c config.yaml -o ./output/

支持的间隔：30s、5m、1h、2h 等。

---

设置向导

通过交互方式生成 YAML 配置：

goldenflow init data.csv

向导会分析您的数据、建议转换，并保存可直接使用的 goldenflow.yaml。

---

历史记录

GoldenFlow 会在 ~/.goldenflow/history/ 中跟踪每次转换运行：

goldenflow history         # Last 20 runs
goldenflow history -n 50   # Last 50 runs

每条运行记录包含：源文件、行数、应用的转换、错误和耗时。

---

架构映射

需要合并来自不同系统的数据时：

goldenflow map --source crm_export.csv --target warehouse_schema.csv

GoldenFlow 会根据名称相似度和数据分析结果，自动映射不同架构中的列：

crm_export.csv              warehouse schema
-----                       -----
email_address      ->       email (rename)
phone_number       ->       phone (rename)
fname              ->       first_name (alias match)
st                 ->       state (alias match)

有歧义的映射会标记出来，供人工审核。确定性高的映射会自动应用。

---

领域套件

适用于常见行业的预配置转换集。目前 5 个套件均已实现：

goldenflow transform patients.csv --domain healthcare
goldenflow transform employees.csv --domain people_hr
goldenflow transform transactions.csv --domain finance
goldenflow transform orders.csv --domain ecommerce
goldenflow transform listings.csv --domain real_estate

领域套件	覆盖内容
人员/人力资源	姓名解析、SSN 格式化、雇佣日期、性别/布尔值标准化
医疗保健	患者 ID、诊断代码、临床日期、HIPAA 敏感字段处理
金融	货币标准化、账号、交易日期、金额解析
电子商务	SKU 标准化、价格解析、订单日期、地址标准化
房地产	房产地址、挂牌日期、价格标准化、地理字段

---

转换库（76 个转换）

文本转换（18 个）

转换	功能
strip	去除首尾空白
lowercase / uppercase	大小写转换
title_case	转换为规范大小写（"john smith" -> "John Smith"）
normalize_unicode	NFKD 标准化，去除重音符号
normalize_quotes	智能/弯引号 -> 直引号
collapse_whitespace	多个空格 -> 单个空格
truncate:N	限制为 N 个字符
remove_punctuation	去除标点符号
remove_html_tags	从抓取的数据中去除 HTML 标记
remove_urls	从自由文本字段中去除 URL
remove_digits	从文本中去除数字字符
remove_emojis	去除表情符号
fix_mojibake	修复常见的 UTF-8/Latin-1 编码乱码
normalize_line_endings	将 \r\n 和 \r 标准化为 \n
extract_numbers	从混合文本中提取数值
pad_left:N / pad_right:N	补齐至固定宽度（账号、ID）

电话转换（5 个）

转换	功能
phone_e164	任意格式 -> +15550123456
phone_national	任意格式 -> (555) 012-3456
phone_digits	仅保留数字
phone_validate	标记无效号码
phone_country_code	提取国家电话区号

姓名转换（8 个）

转换	功能
split_name	"John Smith" -> 名："John"，姓："Smith"
split_name_reverse	"Smith, John" -> 名："John"，姓："Smith"
strip_titles / strip_suffixes	去除 Mr.、Mrs.、Dr.、MD、PhD 等
name_proper	"mcdonald" -> "McDonald"，"o'brien" -> "O'Brien"
initial_expand	标记带有姓名首字母的记录，以供审核
nickname_standardize	"Bob" -> "Robert"，"Bill" -> "William"
merge_name	将 first_name + last_name 合并为 full_name

地址转换（8 个）

转换	功能
address_standardize	"Street" -> "St"，"Avenue" -> "Ave"
address_expand	"St" -> "Street"，"Ave" -> "Avenue"
state_abbreviate / state_expand	"Pennsylvania" <-> "PA"
zip_normalize	补齐前导零、去除 +4、验证
split_address	单行地址 -> 街道、城市、州、邮编
country_standardize	"United States" / "USA" -> "US"（ISO 3166）
unit_normalize	"Apt" / "Apartment" / "#" -> "Unit"

日期转换（13）

转换	功能
date_iso8601	任意格式 -> 2024-03-15
datetime_iso8601	任意格式 -> 2024-03-15T15:30:00
date_us / date_eu	按地区格式输出
date_parse	自动检测并规范化为 ISO 8601
age_from_dob	出生日期 -> 年龄
extract_year / extract_month / extract_day	拆分日期
extract_quarter	日期 -> Q1/Q2/Q3/Q4
extract_day_of_week	日期 -> Monday、Tuesday 等
date_shift	增加/减少天数（匿名化）
date_validate	将无效日期标记为布尔值

分类转换（6）

转换	功能
category_auto_correct	使用模糊匹配将拼写错误匹配为规范值
category_standardize	将不同变体映射为规范值
category_from_file	从 CSV/YAML 文件加载映射
boolean_normalize	"Yes"/"Y"/"1"/"True" -> true
gender_standardize	"Male"/"M"/"Female"/"F" -> M/F
null_standardize	"N/A"/"NULL"/"none" -> null

数值转换（9）

转换	功能
currency_strip	"$1,234.56" -> 1234.56
percentage_normalize	"85%" -> 0.85
round:N	四舍五入到 N 位小数
clamp	将值限制在最小值/最大值范围内
to_integer	将字符串解析为 int，并截断小数
abs_value	绝对值
fill_zero	将 null 替换为 0
comma_decimal	欧洲格式 "1.234,56" -> 1234.56
scientific_to_decimal	"1.5e3" -> 1500.0

电子邮件转换（4）

转换	功能
email_lowercase	规范化为小写
email_normalize	移除 +标签、移除 Gmail 点号并转为小写
email_extract_domain	[email protected] -> example.com
email_validate	标记格式无效的电子邮件地址

标识符转换（3）

转换	功能
ssn_format	规范化为 XXX-XX-XXXX
ssn_mask	隐去为 ***-**-1234
ein_format	规范化为 XX-XXXXXXX

URL 转换（2）

转换	功能
url_normalize	将域名转为小写、确保包含方案并移除末尾斜杠
url_extract_domain	从 URL 中提取域名

---

特殊模式

严格模式

如果任何转换出错，立即失败——适用于 CI 或生产流水线：

goldenflow transform data.csv --strict

如果任何转换失败，将以代码 1 退出，并打印前 5 个错误。

LLM 模式

使用 LLM 改进分类纠正，并处理模糊匹配无法识别的边缘情况：

goldenflow transform data.csv --llm

需要在环境中设置 OPENAI_API_KEY 或 ANTHROPIC_API_KEY。出错时会平稳回退到标准转换。

自动纠正

category_auto_correct 使用模糊匹配自动修正拼写错误的分类值。为避免误报，该功能不会应用于基数较高的列（唯一值占比 >10%）。

"actve" -> "active"
"Pennsylvnia" -> "Pennsylvania"
"Unted States" -> "United States"

---

YAML 配置

用于可重复执行的流水线：

# goldenflow.yaml
source: customers.csv
output: customers_clean.csv

transforms:
  - column: name
    ops: [strip, title_case]
  - column: email
    ops: [lowercase, strip]
  - column: phone
    ops: [phone_e164]
  - column: state
    ops: [state_abbreviate]
  - column: signup_date
    ops: [date_iso8601]

renames:
  email_address: email
  phone_number: phone

drop: [internal_id, temp_notes]

dedup:
  columns: [email]
  keep: first

goldenflow transform customers.csv -c goldenflow.yaml

根据数据自动生成配置：

goldenflow learn data.csv -o config.yaml

---

Python API

import goldenflow

# Zero-config
result = goldenflow.transform_file("messy_data.csv")
print(result.df)          # Clean Polars DataFrame
print(result.manifest)    # Audit trail

# With config
from goldenflow import GoldenFlowConfig, TransformSpec, TransformEngine

config = GoldenFlowConfig(
    transforms=[
        TransformSpec(column="phone", ops=["phone_e164"]),
        TransformSpec(column="date", ops=["date_iso8601"]),
    ]
)
engine = TransformEngine(config=config)
result = engine.transform_df(df)

Jupyter Notebook 支持

TransformResult、Manifest 和 DatasetProfile 都实现了 _repr_html_()——在 Jupyter 中会自动渲染为丰富的 HTML 表格：

import goldenflow

result = goldenflow.transform_file("messy_data.csv")
result           # renders as HTML table in Jupyter
result.manifest  # renders transform audit trail

---

TypeScript / JavaScript

GoldenFlow 提供功能完整、与原版功能对等的 TypeScript 移植版——包含相同的 83 个转换、相同的引擎和相同的配置格式。核心部分可安全用于边缘环境（可在浏览器、Cloudflare Workers、Vercel Edge 中运行），并通过 Node 层处理文件 I/O 和 CLI。

安装

npm install goldenflow

CLI

npx goldenflow-js transform data.csv              # Zero-config
npx goldenflow-js transform data.csv -c config.yaml  # With config
npx goldenflow-js profile data.csv                 # Column profiles
npx goldenflow-js learn data.csv -o config.yaml    # Generate config
npx goldenflow-js diff before.csv after.csv        # Compare files
npx goldenflow-js map -s source.csv -t target.csv  # Schema mapping
npx goldenflow-js stream large.csv --chunk-size 50000  # Streaming
npx goldenflow-js demo                             # Generate sample data
npx goldenflow-js history                          # Show recent runs

TypeScript API

import { TransformEngine, makeConfig } from "goldenflow";

// Zero-config — auto-detect and fix
const engine = new TransformEngine();
const result = engine.transformDf([
  { name: "  John Smith  ", email: "[email protected]", phone: "(555) 123-4567" },
  { name: "DR. JANE DOE", email: "  [email protected]  ", phone: "555.987.6543" },
]);

console.log(result.rows);
// [
//   { name: "John Smith", email: "[email protected]", phone: "+15551234567" },
//   { name: "Jane Doe", email: "[email protected]", phone: "+15559876543" },
// ]
console.log(result.manifest.records.length); // transforms applied

// Configured — explicit transforms per column
const engine = new TransformEngine({
  transforms: [
    { column: "phone", ops: ["phone_e164"] },
    { column: "email", ops: ["strip", "email_normalize"] },
    { column: "name", ops: ["strip", "title_case"] },
    { column: "state", ops: ["state_abbreviate"] },
    { column: "price", ops: ["currency_strip"] },
    { column: "signup_date", ops: ["date_iso8601"] },
  ],
  renames: { email_address: "email" },
  drop: ["internal_id"],
  dedup: { columns: ["email"], keep: "first" },
});
const result = engine.transformDf(rows);

// Schema mapping
import { SchemaMapper } from "goldenflow";

const mapper = new SchemaMapper();
const mappings = mapper.map(
  [{ fname: "John", lname: "Smith", email_address: "[email protected]" }],
  [{ first_name: "", last_name: "", email: "" }],
);
// [{ source: "fname", target: "first_name", confidence: 0.95 }, ...]

// Streaming large datasets
import { StreamProcessor } from "goldenflow";

const processor = new StreamProcessor({ transforms: [{ column: "name", ops: ["strip"] }] });
for (const result of processor.streamRows(largeDataset, 10_000)) {
  await writeChunk(result.rows);
}

// Profiling
import { profileDataframe } from "goldenflow";

const profile = profileDataframe(rows, "customers.csv");
for (const col of profile.columns) {
  console.log(`${col.name}: ${col.inferredType}, ${col.nullCount} nulls, ${col.uniqueCount} unique`);
}

// Edge-safe import (browsers, Workers, Edge Runtime)
import { TransformEngine } from "goldenflow/core";

// Node-only import (includes file I/O, MCP, CLI)
import { readFile, TransformEngine } from "goldenflow/node";

MCP 服务（TypeScript）

import { TOOL_DEFINITIONS, handleTool } from "goldenflow/node";

// TOOL_DEFINITIONS: 10 MCP tools for Claude Desktop
// handleTool("transform", { path: "data.csv" }) → JSON string

REST API（TypeScript）

import { runServer } from "goldenflow/node";
runServer(8000); // Starts HTTP server with /health, /transforms, /transform

---

公共 API（34 个导出）

from goldenflow import (
    # Core engine
    TransformEngine, TransformResult,
    # Config
    GoldenFlowConfig, TransformSpec, SplitSpec, FilterSpec, DedupSpec, MappingSpec,
    # Convenience
    transform_file, transform_df,
    # Manifest
    Manifest, TransformRecord, TransformError,
    # Profiler
    DatasetProfile, ColumnProfile,
    # Selector & differ
    select_transforms, diff_dataframes, DiffResult,
    # Transform registry
    TransformInfo, register_transform, get_transform, list_transforms, parse_transform_name,
    # Mapping
    SchemaMapper, ColumnMapping,
    # Config helpers
    load_config, save_config, merge_configs, learn_config,
    # Domains
    DomainPack, load_domain,
    # Connectors
    read_file, write_file,
)

---

集成

REST API

goldenflow serve --host 0.0.0.0 --port 8000

POST CSV 数据，获取转换后的 CSV。基于 FastAPI 构建。

MCP 服务

goldenflow mcp-serve

将 GoldenFlow 作为 MCP 工具提供给 Claude Desktop。请在 Claude Desktop 设置中进行配置。

TUI

goldenflow interactive data.csv

使用 Textual 构建的功能齐全的终端 UI。浏览配置文件、应用转换并预览结果。

远程 MCP 服务

GoldenFlow 可通过 Smithery 作为托管 MCP 服务使用——无需安装任何内容，即可从任意 MCP 客户端连接。

Claude Desktop / Claude Code：

{
  "mcpServers": {
    "goldenflow": {
      "url": "https://goldenflow-mcp-production.up.railway.app/mcp/"
    }
  }
}

本地服务：

pip install goldenflow[mcp]
goldenflow mcp-serve

提供 10 种工具：转换文件、自动映射架构、分析列、生成配置、比较转换前后的差异，以及应用领域包。

---

Golden Suite 的一部分

工具	用途	Python	TypeScript
GoldenCheck	验证并分析数据质量	pip install goldencheck	npm install goldencheck
GoldenFlow	转换并标准化数据	pip install goldenflow	npm install goldenflow
GoldenMatch	去重并匹配记录	pip install goldenmatch	—
GoldenPipe	编排完整流水线	pip install goldenpipe	—

Raw Data
   |
   v
+--------------+
|  GoldenCheck |  <- Discover quality issues
|  goldencheck |
|  scan data   |
+------+-------+
       | findings
       v
+--------------+
|  GoldenFlow  |  <- Fix issues, standardize, reshape
|  goldenflow  |
|  transform   |
+------+-------+
       | clean data
       v
+--------------+
|  GoldenMatch |  <- Deduplicate, match, create golden records
|  goldenmatch |
|  dedupe      |
+------+-------+
       | golden records
       v
   Clean, deduplicated,
   production-ready data

将它们串联起来：

goldencheck scan data.csv | goldenflow transform --from-findings | goldenmatch dedupe

---

性能

基于 Polars 构建（由 Rust 提供支持的 DataFrame）。转换采用混合方式：对于简单转换（去除空白、转为小写），原生 Polars 表达式会留在 Rust 引擎中执行；对于复杂转换（电话号码解析、日期解析），则通过 map_batches 使用经过优化的 Python。

---

基准测试

在 DQBench 转换基准测试的三个层级（客户数据库、电商、医疗理赔）中，GoldenFlow 均获得 100/100 的成绩。

pip install dqbench
dqbench run goldenflow

---

为什么选择 GoldenFlow？

	GoldenFlow	pandas 脚本	Great Expectations	dbt	Dataprep.Clean
零配置转换	支持（自动检测）	不支持	不支持（仅验证）	不支持（SQL 转换）	部分支持
76 种内置转换（11 个类别）	支持	手动	不支持（验证器，非转换器）	通过 SQL	约 30 种清理工具
领域包（医疗、金融等）	内置 5 种	不支持	不支持	不支持	不支持
架构映射	自动 + 手动	手动	不支持	通过 ref/source	不支持
审计跟踪（清单）	自动生成 JSON	手动	不支持	通过日志	不支持
流式处理 / 大文件	内置	手动分块	不支持	支持（数据仓库）	不支持
MCP 服务	支持	不支持	不支持	不支持	不支持
原生支持 Polars	支持	不支持（pandas）	不支持（pandas/Spark）	不支持（SQL）	不支持（pandas）
DQBench 转换评分	100/100	不适用	不适用	不适用	不适用

GoldenFlow 专为验证与匹配之间的转换步骤打造，并非通用 ETL 工具。它能自动将杂乱数据转换为干净、标准化的数据。

---

错误处理

GoldenFlow 会在 CLI 边界捕获错误，并显示友好且可执行的提示，不会输出原始堆栈跟踪。单个转换错误会记录在清单中，而不会导致运行崩溃。使用 --strict 可更改此行为。

---

进度条

长时间运行的操作（流式处理、监视模式、调度）会显示 Rich 进度旋转指示器，其中包含批次数、已处理行数和预计完成时间。

---

GitHub： github.com/benzsevern/goldenflow 作者： Ben Severn 许可证： MIT Python： 3.11+ | Node.js： 20+ | npm： goldenflow | PyPI： goldenflow