https://flink.apache.org/ logo
Join Slack
Powered by
Hello! I’m trying to create an Iceberg sink in my ...
# troubleshooting
j
Hello! I’m trying to create an Iceberg sink in my Flink application and I’m able to successfully insert some dummy data into the Iceberg table with this simple line of code:
Copy code
tableEnv.executeSql("INSERT INTO flink_catalogue.flink_demo.flink_iceberg_sink VALUES (1, 1, 'dummy_merchant1', 1, '2022-04-01 00:00:00'), (2, 2, 'dummy_merchant2', 2, '2022-04-02 00:00:00');")
but when I try and use the data stream from my Kafka source it’s not writing any data and there don’t seem to be any useful error messages in the stacktrace. Here is the INSERT INTO statement (note, 
flink_transaction
is a kafka-backed Flink source table):
Copy code
tableEnv.executeSql("""
    INSERT INTO flink_catalogue.flink_demo.flink_iceberg_sink
    SELECT
        transaction_id
        , transaction_amount
        , merchant_name
        , user_id
        , created_at
    FROM flink_transaction
    """)
I also checked if the data stream is producing the data I need in the right format by removing the 
INSERT INTO
statement and it did print the results that I was expecting. Can anyone help me understand what I’m doing wrong here please? (I’ll add the full code in-thread)
✅ 1
Copy code
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    // register Flink source table
    tableEnv.executeSql("""
    CREATE TABLE flink_transaction (
        transaction_id INT,
        transaction_amount FLOAT,
        merchant_name VARCHAR,
        user_id INTEGER,
        created_at VARCHAR
    )
    WITH (
        'connector'= 'kafka',
        'topic' = 'transaction',
        'properties.bootstrap.servers' = 'localhost:29092',
        'value.format' = 'avro-confluent',
        'value.avro-confluent.url' = '<http://localhost:8083>',
        'properties.group.id' = 'group-1',
        'scan.startup.mode' = 'earliest-offset'
    );
    """)

    // register Flink sink table
    tableEnv.executeSql("""
    CREATE TABLE flink_iceberg_sink (
        transaction_id INT,
        transaction_amount FLOAT, 
        merchant_name VARCHAR,
        user_id INT,
        created_at VARCHAR
    )
    WITH (
    'connector' = 'iceberg',
    'warehouse' = 's3://<s3_path>',
    'catalog-impl' = 'org.apache.iceberg.aws.glue.GlueCatalog',
    'lock-impl' = 'org.apache.iceberg.aws.glue.DynamoLockManager',
    'lock.table' = 'FlinkGlue4IcebergLockTable'
    );
    """)

    // create a Flink catalogue
    tableEnv.executeSql(
        """
        CREATE CATALOG flink_catalogue WITH (
        'type'='iceberg',
        'warehouse'='<s3_path>',
        'catalog-impl'='org.apache.iceberg.aws.glue.GlueCatalog',
        'io-impl'='org.apache.iceberg.aws.s3.S3FileIO',
        'lock-impl'='org.apache.iceberg.aws.glue.DynamoLockManager',
        'lock.table'='FlinkGlue4IcebergLockTable'
        );
    """)

    tableEnv.executeSql("""
    CREATE TABLE flink_catalogue.flink_demo.flink_iceberg_sink (
        transaction_id INTEGER,
        transaction_amount FLOAT,
        merchant_name VARCHAR,
        user_id INTEGER,
        created_at VARCHAR
    );
    """)

    tableEnv.executeSql("""
    INSERT INTO flink_catalogue.flink_demo.flink_iceberg_sink
    SELECT
        transaction_id
        , transaction_amount
        , merchant_name
        , user_id
        , created_at
    FROM flink_transaction
    """)
The statements above are defined using Kotlin
m
The difference is is that your dummy data is a bounded source, while your Kafka source is an unbounded stream. I'm suspecting that for your unbounded stream, you'll need to enable checkpointing in order for the data to be written to iceberg
j
Ahh yes thanks so much, that makes sense. That is (partially) working now that I’ve added checkpointing. The weird thing is it has committed 1 data file now from the first checkpoint, but it doesn’t commit any more data files on subsequent checkpoints. The metadata files in s3 are showing multiple snapshots and multiple data files though.
Okay ignore that, it is working!.Thanks @Martijn Visser! I’ve been scratching my head for weeks with this 😅
m
Happy to have helped 😅 🫣
Given I have 0 experience with Iceberg, I’m happy that my guess was correct
😄 1
2 Views
Open in Slack
PreviousNext
Powered by