Automate Kafka Integration Tasks from PowerShell

The CData ADO.NET Provider for Apache Kafka is a standard ADO.NET Provider that make it easy to accomplish data cleansing, normalization, backup, and other integration tasks by enabling real-time and bidirectional access to Kafka.

ADO.NET Provider

The ADO.NET Provider provides a SQL interface for Kafka; this tutorial shows how to use the Provider to create, retrieve, update, and delete Kafka data.

Once you have acquired the necessary connection properties, accessing Kafka data in PowerShell can be enabled in three steps.

Set BootstrapServers and the Topic properties to specify the address of your Apache Kafka server, as well as the topic you would like to interact with.

Authorization Mechanisms

• SASL Plain: The User and Password properties should be specified. AuthScheme should be set to 'Plain'.
• SASL SSL: The User and Password properties should be specified. AuthScheme should be set to 'Scram'. UseSSL should be set to true.
• SSL: The SSLCert and SSLCertPassword properties should be specified. UseSSL should be set to true.
• Kerberos: The User and Password properties should be specified. AuthScheme should be set to 'Kerberos'.

You may be required to trust the server certificate. In such cases, specify the TrustStorePath and the TrustStorePassword if necessary.

1. Load the provider's assembly:

   [Reflection.Assembly]::LoadFile("C:\Program Files\CData\CData ADO.NET Provider for Apache Kafka\lib\System.Data.CData.ApacheKafka.dll")
    

2. Connect to Kafka:

    
   $conn= New-Object System.Data.CData.ApacheKafka.ApacheKafkaConnection("User=admin;Password=pass;BootStrapServers=https://localhost:9091;Topic=MyTopic;")
   $conn.Open()
   

3. Instantiate the ApacheKafkaDataAdapter, execute an SQL query, and output the results:

   $sql="SELECT Id, Column1 from SampleTable_1"
   
   $da= New-Object System.Data.CData.ApacheKafka.ApacheKafkaDataAdapter($sql, $conn)
   $dt= New-Object System.Data.DataTable
   $da.Fill($dt)
   
   $dt.Rows | foreach {
   	Write-Host $_.id $_.column1
   }
    

Update Kafka Data

$cmd = New-Object System.Data.CData.ApacheKafka.ApacheKafkaCommand("UPDATE SampleTable_1 SET Column2='100' WHERE Id = @myId", $conn)
$cmd.Parameters.Add((New-Object System.Data.CData.ApacheKafka.ApacheKafkaParameter("@myId","10456255-0015501366")))
$cmd.ExecuteNonQuery()

Insert Kafka Data

$cmd = New-Object System.Data.CData.ApacheKafka.ApacheKafkaCommand("INSERT INTO SampleTable_1 (Column2) VALUES (@myColumn2)", $conn)
$cmd.Parameters.Add((New-Object System.Data.CData.ApacheKafka.ApacheKafkaParameter("@myColumn2","100")))
$cmd.ExecuteNonQuery()

Delete Kafka Data

$cmd = New-Object System.Data.CData.ApacheKafka.ApacheKafkaCommand("DELETE FROM SampleTable_1 WHERE Id=@myId", $conn)
$cmd.Parameters.Add((New-Object System.Data.CData.ApacheKafka.ApacheKafkaParameter("@myId","001d000000YBRseAAH")))
$cmd.ExecuteNonQuery()