Can't write hudi format in catalog

[ltuantai95]

I have code stream from datagen and write hudi format with a catalog

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings
import os

# Set up the streaming execution environment
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)  # good for local testing
env.enable_checkpointing(60000)  # Checkpoint every 60 seconds

# Create table environment for streaming mode (no Blink planner)
t_env = StreamTableEnvironment.create(env)

t_env.execute_sql(
    """
        CREATE CATALOG hudi_catalog WITH (
            'type'         = 'hudi',
            'catalog.path' = 'hdfs://namenode:8020/hudi_catalog'
        )
    """
)
t_env.execute_sql("USE CATALOG hudi_catalog")
t_env.execute_sql("CREATE DATABASE IF NOT EXISTS hudi_db")
t_env.execute_sql("""
    CREATE TABLE IF NOT EXISTS hudi_catalog.hudi_db.orders_datagen (
        order_number BIGINT PRIMARY KEY NOT ENFORCED,
        price        DECIMAL(10,2)
    ) WITH (
        'connector'     = 'datagen',
        'rows-per-second' = '10',
        
        'fields.order_number.kind' = 'sequence',
        'fields.order_number.start' = '1',
        'fields.order_number.end'   = '1000',
        
        'fields.price.kind' = 'random',
        'fields.price.min'  = '20.00',
        'fields.price.max'  = '1000.00'
        )
""")
t_env.execute_sql(
    """
        CREATE TABLE IF NOT EXISTS hudi_catalog.hudi_db.hudi_orders (
            order_number BIGINT PRIMARY KEY NOT ENFORCED,
            price        DECIMAL(32,2)
        ) WITH (
            'connector' = 'hudi',
            'table.type' = 'COPY_ON_WRITE'
        )
    """
)


t_env.execute_sql("""
INSERT INTO hudi_catalog.hudi_db.hudi_orders
SELECT * FROM hudi_catalog.hudi_db.orders_datagen
""")

Job submit success and finish immediately with no errors but no data (*.parquet) writed.

Thanks for your help!

[danny0405]

did you enable checkpoint, does the source table has dataset there?

[ltuantai95]

I enabled checkoint, source table (orders_dataden) does'nt have dataset.
If i don't use catalog and run code as below, it work OK:

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)  # good for local testing
env.enable_checkpointing(10000)  # Checkpoint every 60 seconds
# Optional: more checkpoint settings (recommended)
# env.get_checkpoint_config().set_checkpoint_timeout(120000)
t_env = StreamTableEnvironment.create(env)
# 1. Create a streaming TableEnvironment

# 2. Create the infinite DataGen source table
t_env.execute_sql("""
CREATE TABLE orders_datagen (
    order_number BIGINT,
    price        DECIMAL(10,2)
) WITH (
    'connector' = 'datagen',
    'rows-per-second' = '10',
    
    'fields.order_number.kind' = 'sequence',
    'fields.order_number.start' = '1',
    'fields.order_number.end'   = '10000',
    
    'fields.price.kind' = 'random',
    'fields.price.min'  = '20.00',
    'fields.price.max'  = '10000.00'
)
""")

# 3. Create the Hudi sink table (streaming upsert supported)
# MOR (MERGE_ON_READ) is usually better for streaming; alternative is COPY_ON_WRITE
t_env.execute_sql("""
CREATE TABLE hudi_sink (
    order_number BIGINT PRIMARY KEY NOT ENFORCED,
    price        DECIMAL(10,2)
) WITH (
    'connector' = 'hudi',
    'path' = 'hdfs://namenode:8020/table1',
    'table.type' = 'COPY_ON_WRITE'
)
""")

# 4. Start continuous streaming insert
# This job will run forever until you cancel it (Ctrl+C or kill the Flink job)
t_env.execute_sql("""
INSERT INTO hudi_sink
SELECT * FROM orders_datagen
""")

Maybe i config catalog wrong ?

[ltuantai95]

Some my logs when working with catalog
jobmanager_log.txt