0

我们正在测试将 flink 用于 IOT 数据处理系统，当我尝试将简单函数映射到流上然后将结果写回 Kafka 时遇到错误。不幸的是，它不提供行号

...
  File "pyflink/fn_execution/coder_impl_fast.pyx", line 555, in pyflink.fn_execution.coder_impl_fast.
FlattenRowCoderImpl._encode_field_simple 
TypeError: an integer is required

当我查找此错误时，我没有找到任何有关 pyflink 的信息，尽管从其他项目中弹出的此错误来看，它们似乎指向 python 版本不匹配。在这种情况下，我可以看到 Flink 在我当前的虚拟环境中正确使用了 python，所以我不确定还要检查什么。我的代码如下。我能够运行一个从 Kafka 读取并打印的简单打印 flink 作业，并且最初插入 Kafka 的数据也是从 flink 作业产生的，所以我知道我至少可以在 Kafka 和 Flink 之间进行通信，进行序列化并正确反序列化。任何有关下一步调查内容的指示都会非常有帮助。

def process_test_data():
    print("processing test data with a flink job")

    env = StreamExecutionEnvironment.get_execution_environment()
    env.add_jars(KAFKA_CONNECTOR_JAR)

    input_type = Types.ROW_NAMED(['id', 'string'], [Types.INT(), Types.STRING()])
    output_type = Types.ROW_NAMED(['id', 'val', 'string'],
            [Types.LONG(), Types.LONG(), Types.STRING()])

    deserialization_schema = JsonRowDeserializationSchema.builder() \
        .type_info(type_info=input_type).build()

    kafka_source = FlinkKafkaConsumer(
        topics=KAFKA_SOURCE_TOPIC,
        deserialization_schema=deserialization_schema,
        properties={
            'bootstrap.servers': KAFKA_SERVER_URL,
            'group.id': KAFKA_SOURCE_GROUP
            })
    kafka_source.set_start_from_earliest()

    ds = env.add_source(kafka_source)

    def sample_task(row):
        return Row(id=row['id'], val=row['id'] * 1000, string=row['string']+'_modified')

    ds = ds.map(sample_task, output_type=output_type) \
            .key_by(lambda row: row['id'])

    # Stream the results back to a Kafka topic
    serialization_schema = JsonRowSerializationSchema.builder() \
            .with_type_info(type_info=output_type).build()
    #serialization_schema = SimpleStringSchema()

    kafka_sink = FlinkKafkaProducer(
        topic=KAFKA_SOURCE_TOPIC,
        serialization_schema=serialization_schema,
        producer_config={
            'bootstrap.servers': KAFKA_SERVER_URL,
            'group.id': KAFKA_SOURCE_GROUP
            })
    ds.add_sink(kafka_sink)

    env.execute('datastream_kafka_demo')
    print("data processing complete")
4

0 回答 0