区块链数据上链前，我在Spark里给它做了一次“全身SPA”
去年Q4，我司突然接到一个“创新试点项目”——要把部分核心业务的交易流水写入区块链。产品经理在会上说得云淡风轻：“就是上个链嘛，技术上应该不难吧？”
我当时正用MacBook Pro跑着一个Spark作业，听见这话差点把咖啡喷到键盘上。不难？你怕是没看过我们每天处理20亿条日志的集群负载！
我是老张，在这组干了快两年的大数据开发，日常和Spark、HDFS、Kafka打交道，Windows只用来测试（别问，问就是公司安全策略强制装的）。说实话，刚听到“区块链”三个字时，我第一反应是：这玩意儿不是炒币的吗？跟我们搞批处理的有啥关系？
但现实很骨感——领导说这是战略方向，必须支持。于是，我和组里的兄弟们硬着头皮开始了这场“传统大数据 + 区块链”的综合技术融合实验。
问题比想象中复杂得多
一开始，我以为只是简单地把清洗好的数据扔给区块链节点就完事了。结果一调研才发现：区块链对写入数据有严苛要求——格式固定、不可变、需签名、吞吐有限。而我们的原始数据呢？来自几十个微服务，字段五花八门，还有大量null值和脏数据。
更坑的是，区块链网络（我们用的是Hyperledger Fabric）的TPS（每秒事务数）实测只有300左右。而我们Spark作业每分钟产出的数据量轻松突破50万条。直接对接？分分钟把区块链节点打崩。
“你们能不能先在链下把数据整干净了再给我？” —— 区块链组的同事在钉钉群里发了个哭笑表情。
行吧，那我们就当一回“数据美容师”。
在Spark里给数据做“预处理SPA”
既然不能直接上链，那就得在上链前做足功夫。我们的目标很明确：把海量、杂乱、高吞吐的业务数据，压缩、校验、标准化成低频、结构化、可审计的区块友好格式。
第一步：字段裁剪 + 类型统一
原始交易日志包含用户ID、设备信息、地理位置、商品SKU、价格、优惠券、支付渠道……足足87个字段。但区块链只需要：交易ID、时间戳、金额、参与方公钥、哈希摘要。
于是，我们在Spark SQL里写了个极度精简的SELECT：
SELECT
tx_id,
event_time,
amount,
user_pubkey,
merchant_pubkey,
sha256(concat_ws('|', tx_id, amount, user_pubkey)) AS data_hash
FROM raw_transactions
WHERE event_time >= '${start_time}'
AND status = 'SUCCESS'
注意那个sha256——这是为了后续做链上-链下数据一致性校验。一旦链上记录的hash和我们存的原始数据算出来不一样，说明要么链被篡改，要么我们数据出错了。
第二步：批量聚合，降低上链频率
单条上链太奢侈，我们改成按5分钟窗口聚合，把同一时间窗内的交易打包成一个“批次凭证”，只上链这个批次的元数据 + 所有交易的Merkle Root。
from pyspark.sql import functions as F
batched = (
df
.withColumn("window", F.window("event_time", "5 minutes"))
.groupBy("window")
.agg(
F.count("*").alias("tx_count"),
F.sum("amount").alias("total_amount"),
F.collect_list("data_hash").alias("hashes")
)
.withColumn("merkle_root", compute_merkle_udf("hashes")) # 自定义UDF
)
这样一来，原本每分钟50万条的写入压力，骤降到每5分钟1条区块记录。区块链组的老哥看到监控图都惊了：“你们怎么做到的？我们节点CPU终于不用99%了！”
第三步：失败重试 + 幂等保障
上链操作可能失败（网络抖动、节点超时、签名错误）。我们设计了一个带状态的重试机制：
- 每次成功上链后，将批次ID写入HBase的
blockchain_commit_log表 - 下次作业启动时，先查这张表，跳过已提交的批次
- 重试最多3次，仍失败则告警并人工介入
这招虽然土，但在双11大促期间救了我们好几次。有一次因为Fabric CA证书轮换，导致签名全部失败，要不是有幂等机制，差点重复上链几千条，那可真是“区块链事故”了。
综合权衡：性能、成本与可维护性
在整个方案落地过程中，我们反复在几个维度做取舍：
| 维度 | 初期方案 | 最终方案 | 原因 |
|---|---|---|---|
| 数据粒度 | 单条上链 | 5分钟批次 | TPS限制+Gas成本（私有链也有资源开销） |
| 存储位置 | 链上存全量 | 链上存Hash，链下存明细 | 节省链存储，符合“链上验证、链下存储”最佳实践 |
| 计算引擎 | 新建Flink流任务 | 复用现有Spark批作业 | 团队熟悉度高，运维成本低 |
| 错误处理 | 简单重试 | 状态记录 + 告警 | 避免重复/漏写，满足审计要求 |
说实话，如果从零开始，或许用Flink做实时上链更优雅。但我们团队的主力栈是Spark，而且大部分数据源本身就是T+1的离线表。技术选型不是比谁更时髦，而是看谁能用现有武器打赢仗。
效果：从“不可能”到“稳如老狗”
上线三个月后，我们这套“Spark预处理 + 区块链轻量上链”的架构已经稳定运行。每天处理约3亿条原始交易，最终生成约288个区块记录（24小时 * 12个5分钟窗口），区块链节点负载常年低于30%。
更妙的是，审计部门爱死了这个设计——他们只需要验证链上的Merkle Root，就能确认某段时间内所有交易未被篡改。而我们大数据平台依然保持高吞吐、低成本的优势。
上周五晚上，我还在加班调一个数据倾斜问题，突然收到区块链组的消息：“老张，你们那个批次上链脚本能开源不？隔壁金融组也想用。”
我笑了笑，回了个：“代码可以给，但请我喝杯瑞幸再说。”
一点心得：别被“新技术”吓住
回顾这次经历，最大的感悟是：所谓“综合技术实践”，本质上就是把不同领域的工具拼成一套解决问题的组合拳。区块链不是银弹，Spark也不是万能胶。关键在于理解各自边界，找到衔接点。
很多同学一听“区块链”就想到去中心化、智能合约、DeFi……但其实在企业级场景里，它更像一个高可信的审计日志系统。而我们大数据工程师的角色，就是帮它“减负”——把脏活累活在链下干完，只把最关键的信息交给它保管。
最后吐槽一句：下次产品经理再说“技术上应该不难吧”，我一定要反问：“那你来写Spark作业还是调Fabric SDK？”
——毕竟，代码不会骗人，但需求文档会。
评论 0