DEBBUG
input_table|flink-py-input-stream|eu-west-1|LATEST
output_table|flink-py-output-stream|eu-west-1
INSERT INTO output_table SELECT * FROM input_table
Traceback (most recent call last):
File "/tmp/flink-web-9eabf686-06d5-413a-9e18-2b9e1c5443fd/flink-web-upload/fa66cb4c-47d1-4cf4-ac3e-abf70c7c864d_code/main.py", line 137, in <module>
main()
File "/tmp/flink-web-9eabf686-06d5-413a-9e18-2b9e1c5443fd/flink-web-upload/fa66cb4c-47d1-4cf4-ac3e-abf70c7c864d_code/main.py", line 129, in main
table_env.execute_sql(insert_sql)
File "/opt/amazon/opt/python/pyflink.zip/pyflink/table/table_environment.py", line 828, in execute_sql
File "/opt/amazon/opt/python/py4j-0.10.9.3-src.zip/py4j/java_gateway.py", line 1321, in __call__
File "/opt/amazon/opt/python/pyflink.zip/pyflink/util/exceptions.py", line 158, in deco
pyflink.util.exceptions.TableException: org.apache.flink.table.api.TableException: Failed to execute sql
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:821)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:784)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:861)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:701)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.flink.util.FlinkException: Failed to execute job 'insert-into_default_catalog.default_database.output_table'.
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2108)
at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:188)
at org.apache.flink.table.planner.delegation.DefaultExecutor.executeAsync(DefaultExecutor.java:95)
at org.apache.flink.table.executor.python.ChainingOptimizingExecutor.executeAsync(ChainingOptimizingExecutor.java:73)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:804)
... 14 more
Caused by: java.lang.RuntimeException: org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster.
at org.apache.flink.util.ExceptionUtils.rethrow(ExceptionUtils.java:319)
at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedFunction$2(FunctionUtils.java:75)
at java.base/java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:642)

def create_source_table(
    table_name,
    stream_name,
    region,
    stream_initpos
):
    return """ CREATE TABLE {0} (
        _data STRING,
        _metadata ROW<
            id STRING,
            event_timestamp TIMESTAMP(3)
        >,
        event_time AS _metadata.event_timestamp,
        WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
    )
    WITH (
        'connector' = 'kinesis',
        'stream' = '{1}',
        'aws.region' = '{2}',
        'scan.stream.initpos' = '{3}',
        'format' = 'json',
        'json.timestamp-format.standard' = 'ISO-8601'
    ) """.format(
        table_name, stream_name, region, stream_initpos)


def create_output_table(table_name, stream_name, region):
    return """ CREATE TABLE {0} (
        _data STRING,
        _metadata ROW<
            id STRING,
            event_timestamp TIMESTAMP(3)
        >,
        event_time TIMESTAMP(3),
        WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
    )
    WITH (
        'connector' = 'kinesis',
        'stream' = '{1}',
        'aws.region' = '{2}',
        'sink.batch.max-size' = '1000',
        'format' = 'json',
        'json.timestamp-format.standard' = 'ISO-8601'
    ) """.format(table_name, stream_name, region)

# 4. Inserts the source table data into the sink table
    insert_sql = "INSERT INTO {0} SELECT * FROM {1}".format(
        output_table_name, input_table_name)
    print(f"\n\n{insert_sql}\n\n")
    table_env.execute_sql(insert_sql)