如何使用PutCassandraRecord将NiFi统计的文件数、行数存入Cassandra数据库

需求为实现多服务器扩展并行采集文件，每日统计文件行数和文件总数存入Cassandra数据库，数据文件为竖线分隔的结构化文本，现有采集处理链路为GetFile→MergeContent→CountText→ReplaceText→PutCassandraRecord，完成配置后写入Cassandra查询无返回数据，需要修正配置实现正常写入。
现有处理器链路示意图：

原有PutCassandraRecord配置参考：

前置预检查 #

• 确认Cassandra表结构包含合法分区键，且字段名、字段类型和待写入数据完全匹配，示例表结构如下：

CREATE TABLE IF NOT EXISTS file_stats (
    stat_date date PRIMARY KEY,
    file_count int,
    line_count bigint
);

• 确认NiFi节点与Cassandra集群网络连通，Cassandra账号拥有对应keyspace、表的写入权限，keyspace、表名无拼写错误。

上游处理器配置修正 #

• MergeContent处理器：按日统计场景下选择时间窗口合并策略，Minimum Number of Entries设为1，Maximum Number of Entries根据单日文件量调整，合并后每个FlowFile对应单日所有采集文件，处理器会自动生成merge.count属性，对应合并的文件总数。
• CountText处理器：启用Count Lines属性，统计完成后会自动给FlowFile添加text.line.count属性，对应所有文件的总行数。
• ReplaceText处理器：选择完全替换原有内容策略，替换值填写如下JSON内容，确保输出内容为合法的单条JSON结构，字段名和Cassandra表字段完全对应：

{"stat_date": "${now():format('yyyy-MM-dd')}", "file_count": ${merge.count}, "line_count": ${text.line.count}}

PutCassandraRecord核心配置修正 #

• 配置Record Reader为JsonTreeReader，默认配置即可识别单条JSON结构，无需额外配置Schema Registry（复杂场景可额外添加Avro Schema做字段校验）。
• Cassandra连接配置：
  • Contact Points填写Cassandra节点IP，多个节点用逗号分隔，默认端口9042
  • Keyspace填写目标数据库名称
  • Table Name填写目标统计表名
  • Consistency Level单节点Cassandra选ONE，多节点集群选LOCAL_QUORUM，避免副本数不足导致写入失败
  • 开启Cassandra认证的场景下填写正确的Username和Password
• 错误排查配置：将Put Failure Relationship设置为路由到失败，给失败队列绑定PutFile处理器存储写入失败的FlowFile，可快速定位字段类型不匹配、主键重复等具体问题。

验证步骤 #

启动处理器链路，等待一个合并窗口后，在Cassandra执行查询语句即可返回统计数据：

SELECT * FROM file_stats WHERE stat_date = '2024-xx-xx';

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