Zeppelin Python Flink无法打印到控制台

Question

我使用的是Kinesis Data Analytics Studio，它提供了一个齐柏林飞船环境。

非常简单的代码：

%flink.pyflink
from pyflink.common.serialization import JsonRowDeserializationSchema
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FlinkKafkaConsumer
# create env = determine app runs locally or remotely
env = s_env or StreamExecutionEnvironment.get_execution_environment()
env.add_jars("file:///home/ec2-user/flink-sql-connector-kafka_2.12-1.13.5.jar")
# create a kafka consumer
deserialization_schema = JsonRowDeserializationSchema.builder() 
.type_info(type_info=Types.ROW_NAMED(
['id', 'name'], 
[Types.INT(), Types.STRING()])
).build()
kafka_consumer = FlinkKafkaConsumer(
topics='nihao',
deserialization_schema=deserialization_schema,
properties={
'bootstrap.servers': 'kakfa-brokers:9092', 
'group.id': 'group1'
})
kafka_consumer.set_start_from_earliest()
ds = env.add_source(kafka_consumer)
ds.print()
env.execute('job1')

我可以在本地完成这项工作，可以看到控制台生成的更改日志。然而，我无法在齐柏林飞船中得到同样的结果。

还检查了Flink web控制台任务管理器中的STDOUT，什么都没有。

我是不是错过了什么？搜索了好几天都找不到任何东西。

Answer 1

我不能100%确定，但我认为您可能需要一个接收器来开始通过数据流提取数据，您可能会使用附带的打印接收器功能