Open Wanglw6 opened 1 year ago
kafka read 配置中"format_type": "json"，从Kafka消费json数据写入到hdfs过程中解析json失败。
No response
{ "job": { "common": { "job_id": 333, "instance_id": 333, "job_name": "BitSail_kafka_to_hdfs_test", "user_name": "test" }, "reader":{ "connector":{ "connector":{ "bootstrap.servers":"PLAINTEXT://hadoop001:6667,hadoop002:6667,hadoop003:6667", "topic":"test", "startup-mode":"earliest-offset", "group":{ "id":"test_consumer_group" } } }, "child_connector_type":"kafka", "format_type": "json", "columns": [ { "name": "ID", "type": "string" }, { "name": "ACCOUNT_ID", "type": "string" }, { "name": "ACCOUNT_PARENT", "type": "int" }, { "name": "ACCOUNT_DESCRIPTION", "type": "string" }, { "name": "ACCOUNT_TYPE", "type": "string" }, { "name": "ACCOUNT_ROLLUP", "type": "string" }, { "name": "CUSTOM_MEMBERS", "type": "string" } ], "class":"com.bytedance.bitsail.connector.legacy.kafka.source.KafkaSourceFunctionDAGBuilder" }, "writer": { "hdfs": { "replication": 3, "dump_type": "hdfs.dump_type.json", "compression_codec": "none" }, "dump": { "output_dir": "hdfs:///tmp/streaming_file_hdfs/", "format": { "type": "hdfs" } }, "class": "com.bytedance.bitsail.connector.legacy.streamingfile.sink.FileSystemSinkFunctionDAGBuilder", "partition_infos": "[{\"name\":\"date\",\"type\":\"TIME\"},{\"name\":\"hour\",\"type\":\"TIME\"}]", "event_time_fields":"test001", "rolling.inactivity_interval": 600000, "rolling.max_part_size": 134217728 } } }
0.1.0
BitSail Core
What happened
kafka read 配置中"format_type": "json"，从Kafka消费json数据写入到hdfs过程中解析json失败。
What do you expect to happen
No response
How to reproduce
{ "job": { "common": { "job_id": 333, "instance_id": 333, "job_name": "BitSail_kafka_to_hdfs_test", "user_name": "test" },
"reader":{ "connector":{ "connector":{ "bootstrap.servers":"PLAINTEXT://hadoop001:6667,hadoop002:6667,hadoop003:6667", "topic":"test", "startup-mode":"earliest-offset", "group":{ "id":"test_consumer_group" } } }, "child_connector_type":"kafka", "format_type": "json", "columns": [ { "name": "ID", "type": "string" }, { "name": "ACCOUNT_ID", "type": "string" }, { "name": "ACCOUNT_PARENT", "type": "int" }, { "name": "ACCOUNT_DESCRIPTION", "type": "string" }, { "name": "ACCOUNT_TYPE", "type": "string" }, { "name": "ACCOUNT_ROLLUP", "type": "string" }, { "name": "CUSTOM_MEMBERS", "type": "string" } ], "class":"com.bytedance.bitsail.connector.legacy.kafka.source.KafkaSourceFunctionDAGBuilder" }, "writer": { "hdfs": { "replication": 3, "dump_type": "hdfs.dump_type.json", "compression_codec": "none" }, "dump": { "output_dir": "hdfs:///tmp/streaming_file_hdfs/", "format": { "type": "hdfs" } }, "class": "com.bytedance.bitsail.connector.legacy.streamingfile.sink.FileSystemSinkFunctionDAGBuilder", "partition_infos": "[{\"name\":\"date\",\"type\":\"TIME\"},{\"name\":\"hour\",\"type\":\"TIME\"}]", "event_time_fields":"test001", "rolling.inactivity_interval": 600000, "rolling.max_part_size": 134217728 } } }
Build Environment
No response
Execution Environment
No response
BitSail version
0.1.0
BitSail Component or Code Module
BitSail Core
Are you willing to submit PR?
Code of Conduct