augustodn

CREATE TABLE { input_table_name } (
    . . .
    event_time TIMESTAMP_LTZ(3),
    . . .
    )
    PARTITIONED BY (sensor_id)
    WITH (
       . . .
       'format' = 'json',
       'json.timestamp-format.standard' = 'ISO-8601'

augustodn

...
INSERT INTO {output_table_name}
       SELECT
           message_id,
           sensor_id,
           message.temperature AS temperature,
           'High temperature detected' AS alert,
           event_time
       FROM {input_table_name}
       WHERE

augustodn

def create_input_table(table_name, stream_name, region, stream_initpos=None):
    init_pos = stream_initpos if stream_initpos else ""


    return f"""
        CREATE TABLE {table_name} (
            message_id VARCHAR(32),
            sensor_id INTEGER,
            message ROW(
               temperature FLOAT,

augustodn

statement_set = table_env.create_statement_set()
statement_set.add_insert_sql(
    f"""
    INSERT INTO {output_table_name}
    SELECT
       message_id,
       sensor_id,
       message.temperature AS temperature,
       'High temperature detected' AS alert,
       event_time

augustodn

def configure_kafka_sink(server: str, topic_name: str) -> KafkaSink:

    return (
        KafkaSink.builder()
        .set_bootstrap_servers(server)
        .set_record_serializer(
            KafkaRecordSerializationSchema.builder()
            .set_topic(topic_name)
            .set_value_serialization_schema(SimpleStringSchema())
            .build()

augustodn

def configure_postgre_sink(sql_dml: str, type_info: Types) -> JdbcSink:
    """Makes postgres sink initialization. Config params are set in this function."""
    return JdbcSink.sink(
        sql_dml,
        type_info,
        JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .with_url(f"jdbc:postgresql://{POSTGRES_HOST}/flinkdb")
        .with_driver_name("org.postgresql.Driver")
        .with_user_name("flinkuser")
        .with_password("flinkpassword")

augustodn

flink-jobmanager:
  build:
    context: ./pyflink
    dockerfile: Dockerfile
  ports:
    - "8081:8081"
  environment:
    - |
      FLINK_PROPERTIES=
      jobmanager.rpc.address: flink-jobmanager

augustodn

FROM flink:1.18.0

# Install Python
RUN apt-get update -y && \
apt-get install -y python3 python3-pip python3-dev && \
rm -rf /var/lib/apt/lists/*

# Ensure python command points to python3
RUN ln -s /usr/bin/python3 /usr/bin/python

augustodn

KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:19092,PLAINTEXT_HOST://localhost:9092