Debug Hologres DataStream

Step 1: Create an execution environment

Use StreamExecutionEnvironment.getExecutionEnvironment() to get the Flink execution environment:

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

Step 2: Configure connection parameters

Instantiate the Flink Configuration class and set the required connector options. Use uppercase letters for option names and underscores as separators. For the full list of options, see Hologres connector options (VVR 11+).

Required options:

Configuration config = new Configuration();
config.setString(HologresConfigs.ENDPOINT, "<hologres-endpoint>");
config.setString(HologresConfigs.USERNAME, "<username>");
config.setString(HologresConfigs.PASSWORD, "<password>");
config.setString(HologresConfigs.DATABASE, "<database-name>");
config.setString(HologresConfigs.TABLE, "<table-name>");

Optional: enable CDC (binlog)

Set BINLOG to true to enable a seamless transition from full snapshot consumption to incremental change reading:

config.set(HologresConfigs.BINLOG, true);

Control the changelog output mode using BINLOG_CHANGE_LOG_MODE:

config.set(HologresConfigs.BINLOG_CHANGE_LOG_MODE, BinlogChangeLogMode.ALL);

Convert the Configuration instance to HologresConnectionParam:

HologresConnectionParam param = new HologresConnectionParam(config);

Step 3: Define the source schema

Use TableSchema.builder() to describe the structure of the Hologres table:

TableSchema schema = TableSchema.builder()
    .field("id", DataTypes.BIGINT())
    .field("name", DataTypes.STRING())
    .build();

Step 4: Build the HologresBinlogSource

Construct a HologresBinlogSource with the connection parameters, schema, and read behavior:

HologresBinlogSource source = new HologresBinlogSource(
    param,                        // Connection parameters
    schema,                       // Table schema
    config,                       // Configuration
    StartupMode.INITIAL,          // Start from the beginning; adjust to control the start offset
    "<table-name>",               // Hologres table name
    "",                           // Predicate pushdown filter; leave blank to scan all rows
    -1,                           // Maximum records to read; -1 means no limit
    Collections.emptySet(),       // Values treated as NULL
    Collections.emptyList()       // Metadata columns to carry
);

Step 5: Add the source to the operator chain

Wire the source into the DataStream pipeline and start the job:

env.fromSource(source, WatermarkStrategy.noWatermarks(), "Holo source").print();
env.execute();

Step 1: Create an execution environment and configure connection parameters

Follow Step 1 and Step 2 in the Hologres source section above.

Step 2: Define the dimension table schema

TableSchema dimSchema = TableSchema.builder()
    .field("id", DataTypes.INT().notNull())
    .field("number", DataTypes.BIGINT())
    .primaryKey("id")
    .build();

Step 3: Build the HologresLookupFunction

Create the lookup function using a Hologres reader and cache strategy:

// Retrieve the Hologres table schema
HologresTableSchema tableSchema = HologresTableSchema.get(hologresConnectionParam);

// Configure caching. Supported values: NONE, LRU, ALL
CacheConfig cacheConfig = CacheConfig.createCacheConfig(config, "NONE");

// Use the table's primary keys as lookup keys
String[] lookupKeys = hologresTableSchema.get().getPrimaryKeys();

// Create a JDBC-based reader
AbstractHologresReader<RowData> hologresReader =
    HologresJDBCReader.createTableReader(param, dimSchema, lookupKeys, tableSchema);

// Build the lookup function
var lookupFunction = new HologresLookupFunction(
    "dim_table",                         // Table name
    schema,                              // TableSchema
    lookupKeys,                          // Lookup key fields
    cacheConfig.getCacheStrategy(),      // Cache strategy
    hologresReader,                      // Reader
    true
);

Choosing between `HologresLookupFunction` and `HologresAsyncLookupFunction`:

Step 4: Add the lookup function to the operator chain

Register the lookup function with your DataStream operators as needed for your join logic.

Step 1: Create an execution environment and configure connection parameters

Follow Step 1 and Step 2 in the Hologres source section above.

Step 2: Define the sink table schema

TableSchema sinkSchema = TableSchema.builder()
    .field("id", DataTypes.INT().notNull())
    .field("number", DataTypes.BIGINT())
    .primaryKey("id")
    .build();

Step 3: Create the HologresWriter

AbstractHologresWriter<RowData> hologresWriter =
    HologresJDBCWriter.createRowDataWriter(
        param,
        schema,
        HologresTableSchema.get(param),
        new Integer[0]);

Step 4: Build the HologresSinkFunction and attach it to the pipeline

HologresSinkFunction sinkFunction = new HologresSinkFunction(param, hologresWriter);

env.from(...)
   .addSink(sinkFunction);

env.execute();