Integrera stöd för Apache Kafka Connect i Azure Event Hubs

Artikel
05/13/2021
4 minuter för att läsa

Apache Kafka Connect är ett ramverk för att ansluta och importera/exportera data från/till ett externt system som MySQL, HDFS och File System via ett Kafka-kluster. Den här självstudien vägleder dig genom att använda Kafka Connect Framework med Event Hubs.

Varning

Användning av Apache Kafka Connect Framework och dess kopplingar är inte berättigade till produkt support via Microsoft Azure.

Apache Kafka Connect antar att den dynamiska konfigurationen lagras i komprimerade ämnen med annars obegränsad kvarhållning. Azure Event Hubs implementerar inte komprimering som en Service Broker-funktion och bevarar alltid en tidsbaserad gräns för kvarhållning av händelser, med roten från principen att Azure Event Hubs är en real tids händelse strömnings motor och inte en långsiktig data eller konfigurations lager.

Även om det Apache Kafka projektet kan vara bekvämt att blanda dessa roller, tror Azure att sådan information bäst hanteras i en lämplig databas eller konfigurations lagring.

Många Apache Kafka Connect-scenarier fungerar, men de här konceptuella skillnaderna mellan Apache Kafkas och Azures Event Hubss bevarande modeller kan orsaka att vissa konfigurationer inte fungerar som förväntat.

Den här självstudien vägleder dig genom integreringen av Kafka Connect med en Event Hub och distribuerar grundläggande FileStreamSource-och FileStreamSink-anslutningar. Den här funktionen finns för närvarande som en förhandsversion. De här anslutningsapparna är inte avsedda för produktionsanvändning, men de demonstrerar ett Kafka Connect-scenario med slutpunkt till slutpunkt där Azure Event Hubs fungerar som asynkron Kafka-meddelandekö.

Anteckning

Det här exemplet finns på GitHub.

I den här självstudien gör du följande:

Skapa ett Event Hubs-namnområde
Klona exempelprojektet
Konfigurera Kafka Connect för Event Hubs
Köra Kafka Connect
Skapa anslutningsappar

Förutsättningar

För att kunna slutföra den här genomgången behöver du följande:

En Azure-prenumeration. Om du inte har en skapar du ett kostnadsfritt konto.
Git
Linux/MacOS
Kafka-version (version 1.1.1, Scala-version 2.11), som finns på kafka.apache.org
Läs introduktionsartikeln Event Hubs för Apache Kafka

Skapa ett Event Hubs-namnområde

En Event Hubs-namnrymd krävs för att skicka och ta emot från Event Hubs-tjänster. Instruktioner för att skapa ett namn område och en Event Hub finns i skapa en Event Hub . Hämta Event Hubs-anslutningssträngen och fullständigt domännamn (FQDN) för senare användning. Anvisningar finns i avsnittet om att hämta en Event Hubs-anslutningssträng.

Klona exempelprojektet

Klona Azure Event Hubs-lagringsplatsen och navigera till undermappen tutorials/connect:

git clone https://github.com/Azure/azure-event-hubs-for-kafka.git
cd azure-event-hubs-for-kafka/tutorials/connect

Konfigurera Kafka Connect för Event Hubs

Minimal omkonfiguration krävs vid omdirigering av Kafka Connect-dataflöde från Kafka till Event Hubs. Följande exempel med connect-distributed.properties illustrerar hur du konfigurerar Connect för att autentisera och kommunicera med Kafka-slutpunkten i Event Hubs:

bootstrap.servers={YOUR.EVENTHUBS.FQDN}:9093 # e.g. namespace.servicebus.windows.net:9093
group.id=connect-cluster-group

# connect internal topic names, auto-created if not exists
config.storage.topic=connect-cluster-configs
offset.storage.topic=connect-cluster-offsets
status.storage.topic=connect-cluster-status

# internal topic replication factors - auto 3x replication in Azure Storage
config.storage.replication.factor=1
offset.storage.replication.factor=1
status.storage.replication.factor=1

rest.advertised.host.name=connect
offset.flush.interval.ms=10000

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter

internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false

# required EH Kafka security settings
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{YOUR.EVENTHUBS.CONNECTION.STRING}";

producer.security.protocol=SASL_SSL
producer.sasl.mechanism=PLAIN
producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{YOUR.EVENTHUBS.CONNECTION.STRING}";

consumer.security.protocol=SASL_SSL
consumer.sasl.mechanism=PLAIN
consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{YOUR.EVENTHUBS.CONNECTION.STRING}";

plugin.path={KAFKA.DIRECTORY}/libs # path to the libs directory within the Kafka release

Viktigt

Ersätt {YOUR.EVENTHUBS.CONNECTION.STRING} med anslutnings strängen för din Event Hubs-namnrymd. Anvisningar om hur du hämtar anslutnings strängen finns i Hämta en Event Hubs anslutnings sträng. Här är ett exempel på en konfiguration: sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://mynamespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXXXXXXXX";

Köra Kafka Connect

I det här steget startas en Kafka Connect-arbetare lokalt i distribuerat läge med hjälp av Event Hubs för att upprätthålla klustertillståndet.

Spara filen connect-distributed.properties ovan lokalt. Se till att ersätta alla värden inom klammerparenteser.
Navigera till platsen för Kafka-versionen på datorn.
Kör ./bin/connect-distributed.sh /PATH/TO/connect-distributed.properties. REST API för Connect-arbetaren är redo för interaktion när du ser 'INFO Finished starting connectors and tasks'.

Anteckning

Kafka Connect använder Kafka AdminClient API för att automatiskt skapa ämnen med rekommenderade konfigurationer, inklusive komprimering. En snabb kontroll av namnrymden i Azure-portalen visar att Connect-arbetarens interna ämnen har skapats automatiskt.

Kafka Connect, interna ämnen måste använda komprimering. Event Hubs-teamet ansvarar inte för att åtgärda Felaktiga konfigurationer om interna anslutnings ämnen är felaktigt konfigurerade.

Skapa anslutningsappar

Det här avsnittet vägleder dig genom att skapa anslutningsapparna FileStreamSource och FileStreamSink.

Skapa en katalog för indata- och utdatafiler.
```
mkdir ~/connect-quickstart
```
Skapa två filer: en fil med startdata som anslutningsappen FileStreamSource läser från och en annan som anslutningsappen FileStreamSink skriver till.
```
seq 1000 > ~/connect-quickstart/input.txt
touch ~/connect-quickstart/output.txt
```

Skapa en FileStreamSource-anslutningsapp. Se till att ersätta klamrarna med din startkatalogsökväg.

curl -s -X POST -H "Content-Type: application/json" --data '{"name": "file-source","config": {"connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector","tasks.max":"1","topic":"connect-quickstart","file": "{YOUR/HOME/PATH}/connect-quickstart/input.txt"}}' http://localhost:8083/connectors

Du bör se Event Hub connect-quickstart på Event Hubs-instansen när du har kört kommandot ovan.

Kontrollera statusen för källanslutningsappen.
```
curl -s http://localhost:8083/connectors/file-source/status
```
Du kan även välja att använda Service Bus Explorer för att verifiera att händelserna ankommer till ämnet connect-quickstart.

Skapa en FileStreamSink-anslutningsapp. Se även denna gång till att ersätta klamrarna med din startkatalogsökväg.

curl -X POST -H "Content-Type: application/json" --data '{"name": "file-sink", "config": {"connector.class":"org.apache.kafka.connect.file.FileStreamSinkConnector", "tasks.max":"1", "topics":"connect-quickstart", "file": "{YOUR/HOME/PATH}/connect-quickstart/output.txt"}}' http://localhost:8083/connectors

Kontrollera statusen för kanalmottagarens anslutningsapp.

curl -s http://localhost:8083/connectors/file-sink/status

Kontrollera att data har replikerats mellan filer och att data är identiska i båda filerna.

# read the file
cat ~/connect-quickstart/output.txt
# diff the input and output files
diff ~/connect-quickstart/input.txt ~/connect-quickstart/output.txt

Rensa

Kafka Connect skapar Event Hub-ämnen för att lagra konfigurationer, förskjutningar och status som består även när Connect-klustret har stängts. Såvida du inte vill att de ska bestå rekommenderas det att dessa ämnen tas bort. Du kan även ta bort händelsehubben connect-quickstart, som skapades under loppet av den här genomgången.

Nästa steg

Mer information om Event Hubs för Kafka finns i följande artiklar: