In case anyone is interested, here’s what I did (the actual source I’m using is different; the datagen source below is for demonstration purposes only):

env = StreamExecutionEnvironment.get_execution_environment()

st_env = StreamTableEnvironment.create(env)

…

st_env.execute_sql("DROP TABLE IF EXISTS print_sink")

print_sink_ddl = _f_"""

CREATE TABLE IF NOT EXISTS print_sink (

product_id   BIGINT,

order_number BIGINT,

quantity     INT,

price        DECIMAL(32, 2),

buyer        STRING,

order_time   TIMESTAMP(3)

)

WITH (

'connector'= 'print'

)

"""

st_env.execute_sql("DROP TABLE IF EXISTS datagen_source")

source_ddl = _f_"""

CREATE TABLE IF NOT EXISTS datagen_source (

product_id   BIGINT,

order_number BIGINT,

quantity     INT,

price_int    INT,

price        AS CAST(price_int/100.0 AS DECIMAL(32, 2)),

buyer        STRING,

order_time   TIMESTAMP(3)

)

WITH (

'connector'= 'datagen',

'fields.product_id.min' = '1',

'fields.product_id.max' = '99999',

'fields.quantity.min' = '1',

'fields.quantity.max' = '25',

'fields.price_int.min' = '29',

'fields.price_int.max' = '99999999',

'fields.order_number.min' = '1',

'fields.order_number.max' = '9999999999',

'fields.buyer.length' = '15'

)

"""

simple_select_from_source_query = """

SELECT

product_id,

order_number,

quantity,

price,

buyer,

order_time

FROM datagen_source

"""

st_env.execute_sql(source_ddl)

st_env.execute_sql(print_sink_ddl)

_*`source_select_result = st_env.sql_query(simple_select_from_source_query)`*_ _*`res_ds = st_env.to_data_stream(source_select_result)`*_ _*`rebalanced_ds = res_ds.rebalance()`*_ _*`rebalanced_source_table = st_env.from_data_stream(rebalanced_ds)`*_ _*`st_env.create_temporary_view("source_rebalanced", rebalanced_source_table)`*_

print_query = """

INSERT INTO print_sink

SELECT

product_id,

order_number,

quantity,

price,

buyer,

order_time

FROM source_rebalanced

"""

st_env.execute_sql(print_query)

env.execute()