CODEWITHSUNDEEP
pandasapache-kafkaapache-flinkpyflinkflink-table-api
Vidura Mudalige
Vidura Mudalige

Reputation: 822

Converting Flink dynamic table into a Pandas dataframe

I'm using the pyflink table api to read data from Kafka. Now I want to convert the resultant table into a Pandas dataframe. Here is my code,

exec_env = StreamExecutionEnvironment.get_execution_environment()
exec_env.set_parallelism(1)
t_config = TableConfig()
t_env = StreamTableEnvironment \
    .create(exec_env, environment_settings=EnvironmentSettings
            .new_instance()
            .in_streaming_mode()
            .use_blink_planner().build())

INPUT_TABLE = "source"

t_env \
    .connect(  # declare the external system to connect to
    Kafka()
        .version("universal")
        .topic("Rides")
        .start_from_earliest()
        .property("zookeeper.connect", "zookeeper:2181")
        .property("bootstrap.servers", "kafka:9092")) \
    .with_format(  # declare a format for this system
    Json()
        .fail_on_missing_field(True)
        .schema(DataTypes.ROW([
        DataTypes.FIELD("rideId", DataTypes.BIGINT()),
        DataTypes.FIELD("isStart", DataTypes.BOOLEAN()),
        DataTypes.FIELD("eventTime", DataTypes.STRING()),
        DataTypes.FIELD("lon", DataTypes.FLOAT()),
        DataTypes.FIELD("lat", DataTypes.FLOAT()),
        DataTypes.FIELD("psgCnt", DataTypes.INT()),
        DataTypes.FIELD("taxiId", DataTypes.BIGINT())]))) \
    .with_schema(  # declare the schema of the table
    Schema()
        .field("rideId", DataTypes.BIGINT())
        .field("taxiId", DataTypes.BIGINT())
        .field("isStart", DataTypes.BOOLEAN())
        .field("lon", DataTypes.FLOAT())
        .field("lat", DataTypes.FLOAT())
        .field("psgCnt", DataTypes.INT())
        .field("eventTime", DataTypes.STRING())) \
    .in_append_mode() \
    .create_temporary_table(INPUT_TABLE)

table = t_env.from_path(INPUT_TABLE)
df = table.to_pandas()

But here I'm getting neither an error nor a result. I'm using the Flink 1.11.3. Is there a way to convert this dynamic table into a static table or something to make table.to_pandas() work?

Upvotes: 0

Views: 719

Answers (1)

Xingbo Huang
Xingbo Huang

Reputation: 383

We can't call to_pandas in an unbounded stream Table. to_pandas can only be called in a bounded Table.

Upvotes: 3

Related Questions