# 初始化项目
$ cd etl-transformation-tool
$ npm init -y
$ npm install commander chalk

# 创建数据转换工具代码
$ cat > index.js << 'EOF' #!/usr/bin/env node const { program }=require('commander'); const chalk=require('chalk'); const fs=require('fs'); program .version('1.0.0') .description('ETL数据转换工具'); // 数据清洗 program .command('clean

‘)
.description(‘数据清洗’)
.option(‘-r, –remove-nulls’, ‘移除空值’, false)
.option(‘-t, –trim’, ‘去除空格’, false)
.option(‘-d, –dedupe’, ‘去除重复’, false)
.action((input, output, options) => {
try {
console.log(chalk.bold(`\n数据清洗: ${input} -> ${output}\n`));

const data = fs.readFileSync(input, ‘utf8’);
const lines = data.split(‘\n’);
const headers = lines[0].split(‘,’);

let records = [];
for (let i = 1; i < lines.length; i++) { if (lines[i].trim()) { records.push(lines[i].split(',')); } } console.log(chalk.blue(`原始记录数: ${records.length}`)); // 移除空值 if (options.removeNulls) { records=records.filter(record=> {
return !record.some(field => field === ‘NULL’ || field === ”);
});
console.log(chalk.green(`移除空值后记录数: ${records.length}`));
}

// 去除空格
if (options.trim) {
records = records.map(record => {
return record.map(field => field.trim());
});
console.log(chalk.green(‘已去除字段空格’));
}

// 去除重复
if (options.dedupe) {
const uniqueRecords = [];
const seen = new Set();
records.forEach(record => {
const key = record.join(‘,’);
if (!seen.has(key)) {
seen.add(key);
uniqueRecords.push(record);
}
});
records = uniqueRecords;
console.log(chalk.green(`去除重复后记录数: ${records.length}`));
}

// 写入输出文件
const outputData = [headers.join(‘,’), …records.map(r =>
r.join(‘,’))].join(‘\n’);
fs.writeFileSync(output, outputData);
console.log(chalk.green(`\n数据已清洗并保存到: ${output}`));

} catch (error) {
console.error(chalk.red(`数据清洗失败: ${error.message}`));
}
});

// 数据转换
program
.command(‘transform

‘)
.description(‘数据转换’)
.option(‘-f, –format ‘, ‘输出格式’, ‘csv’)
.option(‘-m, –mapping ‘, ‘字段映射文件’, ”)
.action((input, output, options) => {
try {
console.log(chalk.bold(`\n数据转换: ${input} -> ${output}\n`));

const data = fs.readFileSync(input, ‘utf8’);
const lines = data.split(‘\n’);
const headers = lines[0].split(‘,’);

let records = [];
for (let i = 1; i < lines.length; i++) { if (lines[i].trim()) { records.push(lines[i].split(',')); } } console.log(chalk.blue(`记录数: ${records.length}`)); // 字段映射 if (options.mapping) { const mapping=JSON.parse(fs.readFileSync(options.mapping, 'utf8' )); const newHeaders=headers.map(h=> mapping[h] || h);
headers.length = 0;
headers.push(…newHeaders);
console.log(chalk.green(‘字段映射已应用’));
}

// 格式转换
let outputData;
if (options.format === ‘json’) {
const jsonData = records.map(record => {
const obj = {};
headers.forEach((header, index) => {
obj[header] = record[index];
});
return obj;
});
outputData = JSON.stringify(jsonData, null, 2);
console.log(chalk.green(‘已转换为JSON格式’));
} else {
outputData = [headers.join(‘,’), …records.map(r =>
r.join(‘,’))].join(‘\n’);
console.log(chalk.green(‘保持CSV格式’));
}

fs.writeFileSync(output, outputData);
console.log(chalk.green(`\n数据已转换并保存到: ${output}`));

} catch (error) {
console.error(chalk.red(`数据转换失败: ${error.message}`));
}
});

// 数据聚合
program
.command(‘aggregate

‘)
.description(‘数据聚合’)
.option(‘-g, –group-by ‘, ‘分组字段’, ”)
.option(‘-s, –sum ‘, ‘求和字段’, ”)
.option(‘-a, –avg ‘, ‘平均值字段’, ”)
.action((input, output, options) => {
try {
console.log(chalk.bold(`\n数据聚合: ${input}
-> ${output}\n`));

const data = fs.readFileSync(input,
‘utf8’);
const lines = data.split(‘\n’);
const headers = lines[0].split(‘,’);

let records = [];
for (let i = 1; i < lines.length; i++) { if (lines[i].trim()) { records.push(lines[i].split(',')); } } console.log(chalk.blue(`记录数: ${records.length}`)); // 分组聚合 if (options.groupBy) { const groupIndex=headers.indexOf(options.groupBy); const sumIndex=options.sum ? headers.indexOf(options.sum) : -1; const groups={}; records.forEach(record=> {
const key = record[groupIndex];
if (!groups[key]) {
groups[key] = { count: 0, sum: 0 };
}
groups[key].count++;
if (sumIndex >= 0) {
groups[key].sum +=
parseFloat(record[sumIndex]) || 0;
}
});

const aggregatedData =
Object.entries(groups).map(([key,
value]) => {
return
`${key},${value.count},${value.sum.toFixed(2)}`;
});

const outputData =
`${options.groupBy},count,sum\n${aggregatedData.join(‘\n’)}`;
fs.writeFileSync(output,
outputData);
console.log(chalk.green(`\n数据已聚合并保存到:
${output}`));
}

} catch (error) {
console.error(chalk.red(`数据聚合失败:
${error.message}`));
}
});

// 执行命令
program.parse(process.argv);
EOF

# 测试数据转换工具
$ chmod +x index.js
$ ./index.js –version
1.0.0

# 数据清洗
$ ./index.js clean employees.csv
employees_cleaned.csv –remove-nulls
–trim –dedupe

数据清洗: employees.csv ->
employees_cleaned.csv

原始记录数: 100
移除空值后记录数: 95
已去除字段空格
去除重复后记录数: 95

数据已清洗并保存到: employees_cleaned.csv

# 数据转换
$ ./index.js transform
employees_cleaned.csv employees.json
–format json

数据转换: employees_cleaned.csv ->
employees.json

记录数: 95
已转换为JSON格式

数据已转换并保存到: employees.json

# 查看JSON数据
$ head -20 employees.json
[
{
“id”: “1”,
“name”: “风哥1号”,
“department”: “技术部”,
“salary”: “15000.00”,
“hire_date”: “2020-01-15”
},
{
“id”: “2”,
“name”: “风哥2号”,
“department”: “市场部”,
“salary”: “12000.00”,
“hire_date”: “2020-02-20”
},
{
“id”: “3”,
“name”: “王五”,
“department”: “财务部”,
“salary”: “13000.00”,
“hire_date”: “2020-03-10”
}
]

# 数据聚合
$ ./index.js aggregate
employees_cleaned.csv
employees_summary.csv –group-by
department –sum salary

数据聚合: employees_cleaned.csv ->
employees_summary.csv

记录数: 95

数据已聚合并保存到: employees_summary.csv

# 查看聚合数据
$ cat employees_summary.csv
department,count,sum
技术部,30,480000.00
市场部,25,300000.00
财务部,20,260000.00
人事部,20,220000.00

# 初始化项目
$ cd etl-loading-tool
$ npm init -y
$ npm install commander chalk mysql

# 创建数据加载工具代码
$ cat > index.js << 'EOF' #!/usr/bin/env node const { program }=require('commander'); const chalk=require('chalk'); const mysql=require('mysql'); const fs=require('fs'); program .version('1.0.0') .description('ETL数据加载工具'); // 创建数据库连接 function createConnection(config) { return mysql.createConnection({ host: config.host, user: config.user, password: config.password, database: config.database }); } // 加载数据 program .command('load