Hello,
We are currently trying to fetch changes from SQL Server CDC tables and push them into a Kafka topic with Nifi.
Basically the flow is working but we would like to have feedback on how to make it more reliable and optimized.
Here is the SQL used to fetch changes from MY_TABLE from offset 37979520
SELECT
CT.SYS_CHANGE_VERSION,
CASE CT.SYS_CHANGE_OPERATION
WHEN 'D' THEN CT.T_PK
ELSE NULL
END as DeletedPK,
CHANGE_TRACKING_CURRENT_VERSION() as LastTransactionId,
T.*
FROM CHANGETABLE(CHANGES MY_TABLE, 37979520) CT
LEFT JOIN MY_TABLE T on CT.T_PK = T.T_PK
Here is the Nifi implementation:
(bigger picture in attachment)
To avoid threading issues, all processors are targeting the primary node with a single thread.
Here are the steps
1 - GenerateFlowFile
To start the process, I generate a FlowFile with a custom property "increment.value" = cache_key.
2 - FetchDistributedMapCache
I fetch the distributed cache with cache identifier $(increment.value} and I put the result in the "stored.state" attribute.
3 - UpdateAttribute
To manage initialization, I update "stored.state" with the following expression:
${stored.state:isNull():ifElse(0, ${stored.state})}
=> If the cache is impty, I start from 0.
4 - ExecuteSQL
SQL statement to fetch data from the current offset
SELECT
CT.SYS_CHANGE_VERSION,
CASE CT.SYS_CHANGE_OPERATION
WHEN 'D' THEN CT.T_PK
ELSE NULL
END as DeletedPK,
CHANGE_TRACKING_CURRENT_VERSION() as LastTransactionId,
T.*
FROM CHANGETABLE(CHANGES MY_TABLE, ${stored.state}) CT
LEFT JOIN MY_TABLE T on CT.T_PK = T.T_PK
5 - QueryRecord
The "Include Zero Record FlowFiles" is set to false to avoid dealing with incoming empty datasets.
The relation selectLastTransactionId fetch the last offset with,
select max(LastTransactionId) as last_transaction_id from flowfile
The relation selectData forward non-empty FlowFiles to Kafka
select * from flowfile
6A - EvaluateJsonPath
I get $.[0].last_transaction_id to forward the next offet to PutDistributedMapCache processor.
6B - PublishKafkaRecord
7A - PutDistributedMapCache
I update the cache with the new offset.
------------------------
Here are some questions,
- Is there a better way to safely start the process ? A GenerateFlowFile with a long Timer Driver Run Schedule is not ideal.
- Despite the (Include Zero Record FlowFiles = false) of the QueryRecord, I have some activity for empty datasets coming from the ExecuteSQL processor.
It triggers many exceptions at PutDistributedMapCache level,
How is this possible ?
- How can we make this flow more reliable ? Is there a better way to orchestrate such a flow ?
Many thanks !
