http://starrocks.io logo
Join Slack
Powered by
<@U0A71G31CDV> what is the most recommended way co...
# questions-and-troubleshooting
c
@Rocky what is the most recommended way config for starrocks connector if the debezium source will have fields in this format? io.debezium.time.Date
r
When dealing with Debezium source data where date fields are in the 
io.debezium.time.Date
format, the primary challenge is that Debezium represents this type as an integer (representing the number of days since the epoch, 1970-01-01) by default. StarRocks' Stream Load, which the connectors use internally, typically expects the 
DATE
type in a string format like 
YYYY-MM-DD
. Here are the recommended configurations depending on which StarRocks connector you are using: 1. StarRocks Kafka Sink Connector If you are using the Kafka Sink connector, you should handle this at the source or via Single Message Transforms (SMTs) in the Kafka Connect configuration. * Configure the Debezium Source Connector: Set the 
time.precision.mode
to 
connect
. This ensures that Debezium uses the standard Kafka Connect logical types for date and time.
Copy code
json
    "time.precision.mode": "connect"
* Use SMTs in the Sink Connector: Use the 
ExtractNewRecordState
(unwrap) transform to simplify the Debezium record. If the field still arrives as an integer, you can use the 
TimestampConverter
SMT to convert the logical Date type into a string that StarRocks can parse.
Copy code
properties
    transforms=unwrap,convertDate
    transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
    transforms.convertDate.type=org.apache.kafka.connect.transforms.TimestampConverter$Value
    transforms.convertDate.field=your_date_column_name
    transforms.convertDate.target.type=string
    transforms.convertDate.format=yyyy-MM-dd
2. StarRocks Flink Connector If you are using Flink CDC to stream data into StarRocks, Flink generally handles the mapping of 
io.debezium.time.Date
to Flink's internal 
DATE
type (and then to 
YYYY-MM-DD
for StarRocks) automatically. However, if you need to explicitly control the format or handling, you can pass Debezium-specific properties in the 
WITH
clause of your Flink SQL table definition. For example:
Copy code
sql
CREATE TABLE st_sink (
    ...
) WITH (
    'connector' = 'starrocks',
    'sink.properties.format' = 'json',
    -- Passes this property to the underlying Debezium deserializer if applicable
    'debezium.date.format' = 'yyyy-MM-dd'
);
Note: The 
debezium.
prefix in the Flink connector's 
WITH
clause allows you to pass configurations directly to the Debezium engine. 3. StarRocks Spark Connector When using Spark, the connector typically maps StarRocks 
DATE
types to Spark 
DataTypes.DateType
. Since Debezium data is often read as a DataFrame, you should ensure that the Spark schema correctly identifies the integer as a date using:
Copy code
scala
df.withColumn("date_col", col("date_col").cast("date"))
Summary Recommendation The most robust approach is to ensure the Debezium Source is configured with 
"time.precision.mode": "connect"
. This allows the StarRocks Sink connectors to interpret the field as a logical Date rather than a raw integer, ensuring compatibility with StarRocks' 
DATE
column format. References • loading/Spark-connector-starrocks.md • loading/Kafka-connector-starrocks.md • faq/loading/Flink_connector_faq.md • sql-reference/data-types/date-types/DATE.md
c
@Rocky im using StarRocks Kafka Sink Connector
r
That's great! The StarRocks Kafka Sink Connector is a powerful tool for continuously loading data from Kafka topics into StarRocks tables. Whether you are running it in Standalone or Distributed mode, here is a quick overview of the key components to ensure a smooth setup. 1. Key Configuration Parameters In your 
connect-StarRocks-sink.properties
(or via REST API in distributed mode), make sure you have these essentials: * 
connector.class
: Must be 
com.starrocks.connector.kafka.StarRocksSinkConnector
. * 
starrocks.http.url
: The FE HTTP address (default port is 
8030
). * 
starrocks.database.name
: Target database. * 
starrocks.topic2table.map
: Mapping between Kafka topics and StarRocks tables (e.g., 
topic1:table1,topic2:table2
). * 
sink.properties.strip_outer_array
: Usually set to 
true
if you are sending JSON arrays. 2. Handling CDC Data (Debezium) If you are syncing data from databases like MySQL or PostgreSQL using Debezium into a Primary Key table, you'll need to add specific transformations: * 
transforms
: Use 
addfield
and 
unwrap
. * 
transforms.addfield.type
: 
com.starrocks.connector.kafka.transforms.AddOpFieldForDebeziumRecord
(this helps StarRocks understand Upsert/Delete operations). 3. Performance Tuning If you're looking to optimize throughput or latency, keep an eye on these: * 
bufferflush.maxbytes
: Controls how much data is buffered before flushing to StarRocks. * 
bufferflush.intervalms
: Controls the maximum time data stays in the buffer. * 
tasks.max
: Increase this to parallelize consumption if your Kafka topic has multiple partitions. Are you running into any specific errors, or do you need help with a particular data format (like Protobuf or Avro)? Let me know and I can provide more detailed guidance! References • loading/Kafka-connector-starrocks.md
c
I want to properly handle this messages from the source with format io.debezium.time.Date
@Rocky I want to properly handle this messages from the source with format io.debezium.time.Date
r
The 
io.debezium.time.Date
format represents a date as the number of days since the epoch (1970-01-01). In StarRocks, this is typically received as an integer when consuming Debezium JSON messages (e.g., via Routine Load or the Kafka Connector). To properly handle and convert this into a StarRocks 
DATE
type, you can use the following methods depending on your ingestion method: 1. Using Routine Load If you are using Routine Load to consume Debezium JSON, you should point your 
json_root
to the 
after
(or 
before
) field and use a transformation in the 
COLUMNS
clause. Since the value is "days since 1970-01-01", you can use the 
DATE_ADD
function:
Copy code
sql
CREATE ROUTINE LOAD demo_db.debezium_load ON target_table
COLUMNS(
    id,
    -- Assume 'dbz_date' is the field name in the Debezium message
    temp_date, 
    -- Transform the days integer into a StarRocks DATE
    actual_date = DATE_ADD('1970-01-01', INTERVAL temp_date DAY)
)
PROPERTIES (
    "format" = "json",
    "json_root" = "$.after", -- Standard Debezium envelope
    "jsonpaths" = "[\"$.id\", \"$.dbz_date\"]"
)
FROM KAFKA (...);
2. Using StarRocks Kafka Connector (Sink) If you are using the StarRocks Kafka Connector, it is recommended to use the Debezium 
ExtractNewRecordState
SMT (Single Message Transform) to flatten the record. If the date still arrives as an integer, you can handle it in the StarRocks table schema by loading it into a temporary 
INT
column and then using a View or a Task to convert it, or ideally, ensure your Debezium configuration uses 
time.precision.mode=connect
, which may map it to a format the connector recognizes more easily. 3. Other Debezium Time Formats For completeness, if you encounter other Debezium time types, here is how to handle them: * 
io.debezium.time.Timestamp
 (Milliseconds since epoch): Use 
FROM_UNIXTIME(col / 1000)
* 
io.debezium.time.MicroTimestamp
 (Microseconds since epoch): Use 
FROM_UNIXTIME(col / 1000000)
* 
io.debezium.time.NanoTimestamp
 (Nanoseconds since epoch): Use 
FROM_UNIXTIME(col / 1000000000)
Summary Checklist * Data Type: Ensure the target column in StarRocks is defined as 
DATE
. * Formula: Always use 
DATE_ADD('1970-01-01', INTERVAL <value> DAY)
for 
io.debezium.time.Date
. * Timezone: Verify your StarRocks 
time_zone
setting (default is 
Asia/Shanghai
) to ensure the date doesn't shift due to offset boundaries, though for 
DATE
(days), this is rarely an issue compared to 
DATETIME
. References • loading/RoutineLoad.md • sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md • loading/loading_introduction/feature-support-loading-and-unloading.md
Open in Slack
PreviousNext
Powered by