流计算 Formats

Format 用于将 connector 读取的消息解析为 source 的列结构。

支持的 Formats

Format	适用 connector	说明
JSON	Kafka、MQTT、HTTP	适合结构化事件消息
CSV	Kafka、MQTT、HTTP	适合简单表格型文本或按行输入
Parquet	HTTP	适合一次性抓取或轮询返回的完整 Parquet 文件

通用规则

通用配置项

配置项	类型	默认值	必选	说明
format	STRING	无	Yes	指定消息格式，当前支持 json 和 csv
bad_data	STRING	drop	No	坏数据处理策略，支持 drop 或 fail

说明：

• bad_data 仅对 source 生效
• json 和 csv 默认按逐行方式解码，适合 newline-delimited JSON 和按行 CSV
• parquet 按整个 payload 解码为一份完整 Parquet 文件

JSON

JSON 特点

• 适合结构化或半结构化事件
• 常用于 Kafka、MQTT 和 HTTP 返回的结构化消息
• 字段名通常可直接映射到 source 列名

JSON 示例数据

{"ts":"2025-01-01T00:00:01Z","sid":"sid-1","value":1.0}
{"ts":"2025-01-01T00:00:02Z","sid":"sid-2","value":2.0}

JSON 配置项

配置项	类型	默认值	必选	说明
unstructured	BOOL	false	No	是否允许更宽松的 JSON 解码，默认按 schema 严格解析

JSON 配置示例

CREATE SOURCE src_json (
  ts TIMESTAMP(9),
  sid STRING,
  value FLOAT64
) WITH (
  connector='kafka',
  brokers='127.0.0.1:9092',
  topic='topic_json_demo',
  format='json',
  unstructured='false',
  bad_data='fail'
);

CSV

CSV 特点

• 结构简单，适合规则化文本数据
• 适合 HTTP 接口返回的表格型数据，也可用于 Kafka 或 MQTT 中的按行 CSV 消息
• 每行对应一条记录

CSV 示例数据

2025-01-01T00:00:03Z,sid-once,101
2025-01-01T00:00:04Z,sid-poll-1,201

CSV 配置项

配置项	类型	默认值	必选	说明
has_header	BOOL	false	No	是否将首行作为表头
delimiter	STRING	,	No	单字节分隔符

CSV 配置示例

CREATE SOURCE src_csv (
  ts TIMESTAMP(9),
  sid STRING,
  value FLOAT64
) WITH (
  connector='http',
  endpoint='http://127.0.0.1:18080/poll',
  method='GET',
  poll='interval(1000)',
  format='csv',
  has_header='false',
  delimiter=','
);

Parquet

Parquet 特点

• 适合 HTTP connector 返回的完整 Parquet 文件
• 更适合批量导入型 source，而不是逐行文本流
• 压缩信息由 Parquet 文件自身 metadata 描述，无需额外配置

Parquet 配置示例

CREATE SOURCE src_parquet (
  ts TIMESTAMP(9),
  sid STRING,
  value FLOAT64
) WITH (
  connector='http',
  endpoint='http://127.0.0.1:18080/export.parquet',
  method='GET',
  poll='once',
  format='parquet',
  bad_data='fail'
);

选型建议

格式	推荐场景	不足
JSON	结构化事件、MQTT / Kafka 消息	文本体积通常更大
CSV	简单行式数据、HTTP 接口文本返回	字段可读性和扩展性较弱
Parquet	HTTP 返回的列式文件、批量抓取	当前仅支持 HTTP connector

常见问题

bad_data 怎么选择

当上游消息偶尔出现坏行、但更关注链路持续可用时，可以使用 bad_data='drop'，让系统跳过当前坏批次。

当希望格式错误立即暴露、便于定位上游数据问题时，可以使用 bad_data='fail'。

相关文档

• 流计算 Connectors