Apache Pulsar数据导入Apache Pinot REALTIME表失败求助

以下是导致数据无法导入的核心问题及修正方案：

1. Topic名称拼写错误 #

表配置中stream.pulsar.topic.name的值为pulsar.pinot.dmeo，存在拼写错误（dmeo应为demo），直接导致Pinot无法连接到目标Pulsar Topic。

修正后：

"stream.pulsar.topic.name": "pulsar.pinot.demo"

2. 时间字段转换逻辑错误 #

ingestion配置中ts字段的转换函数"timestamp"*1000完全无效：

• "timestamp"是字符串常量，并非实际时间变量
• 你的Pulsar消息内容中没有timestamp字段，只有Pulsar自带的publishTime元数据

修正方案：使用Pinot内置的PULSAR_PUBLISH_TIME()函数获取消息的发布时间，转换为毫秒级时间戳：

{
    "columnName": "ts",
    "transformFunction": "PULSAR_PUBLISH_TIME()"
}

3. 冗余的JSON字段转换配置 #

你已经配置了JSONMessageDecoder，该解码器会自动将Pulsar消息的content字段解析为JSON结构并映射到Schema对应的字段。此时再用JSONPATH(content, '$.username')做转换属于冗余操作，反而可能导致字段映射失败。

修正方案：移除ingestionConfig中所有针对username和password的transformConfigs配置，让JSONMessageDecoder自动完成字段映射。

修正后的完整表配置 #

{
    "tableName": "pinot_pulsar_demo",
    "tableType": "REALTIME",
    "segmentsConfig": {
        "timeColumnName": "ts",
        "timeType": "MILLISECONDS",
        "schemaName": "pinot_pulsar_demo",
        "replicasPerPartition": "1"
    },
    "tenants": {},
    "tableIndexConfig": {
        "loadMode": "MMAP",
        "streamConfigs": {
            "streamType": "pulsar",
            "stream.pulsar.topic.name": "pulsar.pinot.demo",
            "stream.pulsar.bootstrap.servers": "pulsar://broker:6650",
            "stream.pulsar.consumer.prop.auto.offset.reset": "smallest",
            "stream.pulsar.consumer.type": "lowlevel",
            "stream.pulsar.fetch.timeout.millis": "30000",
            "stream.pulsar.decoder.class.name": "org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder",
            "stream.pulsar.consumer.factory.class.name": "org.apache.pinot.plugin.stream.pulsar.PulsarConsumerFactory",
            "realtime.segment.flush.threshold.rows": "0",
            "realtime.segment.flush.threshold.time": "1m",
            "realtime.segment.flush.threshold.segment.size": "1M"
        }
    },
    "metadata": {
        "customConfigs": {}
    },
    "ingestionConfig": {
        "transformConfigs": [
            {
                "columnName": "ts",
                "transformFunction": "PULSAR_PUBLISH_TIME()"
            }
        ]
    }
}

额外检查项 #

• 确认Pulsar broker地址pulsar://broker:6650可被Pinot集群访问
• 等待至少1分钟（对应realtime.segment.flush.threshold.time配置），或发送足够数据触发segment flush
• 查看Pinot server和controller的日志，排查是否有连接错误或解析异常

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