Howto konfigurieren kafkaConnect von kafka zu cassandra

Question

ich einrichten möchten verbinden die kafka von kafka Thema cassandra

Das Problem ist einfach: sagen, dass ich wie ein Demo-Thema

in kafka mit json Daten

{"id":"1", "name":"Alex", "clicks":2} 

Ich möchte es automatisch in die Cassanra-Tabelle mit Spalten ID, Name, Klicks schieben.

Ich suche nach kafka-connect-cassandra, aber das einzige Beispiel, das ich finden kann, ist das Lesen von Cassandra und Schreiben zu einem anderen Cassandra-Tisch über Kafka in der Mitte.

Meine Frage ist, wie kann ich es von Kafka und nicht von Cassandra lesen lassen? Ich suche nach einem Connector Open Source mit Beispiel dafür.

Answer 1

Das Beispiel, auf das Sie sich beziehen, zeigt die Source- und Sink-Eigenschaften des Steckers zusammen. Wenn Ihr Anwendungsfall darin besteht, Daten vom Kafka-Thema in eine Cassandra-Tabelle zu übertragen, benötigen Sie lediglich eine Senke. Befolgen Sie diese Schritte

1. Erstellen Sie Ihre eigene Senke Eigenschaftendatei. Verwenden Sie this as an Beispiel. Speichern Sie es als my-sink.properties
2. Gehen Sie auf die Home-Verzeichnis der Installation und führen Sie den Befehl CLASSPATH=<<path-to-connector-jar>> ./bin/connect-standalone connect-standalone.properties my-sink.properties

Wenn Sie in Beispiel detailliertere Schritte interessiert sind, finden Sie hier: https://github.com/yaravind/kafka-connect-jenkins#standalone-mode (Full Disclosure: Ich behaupte, dass die Stecker für Jenkins .)

Answer 2

Ich hatte das gleiche Problem, und ich folgte, was auf https://www.confluent.io/blog/kafka-connect-cassandra-sink-the-perfect-match/ ist. Ich verwende den DataMountaineer-Treiber (http://docs.datamountaineer.com/en/latest/cassandra-sink.html) und stelle ihn im verteilten Modus ein.

Sobald Sie haben es festgelegt, Ihre Konfiguration Json für den cassandra-Anschluss (über REST-API für konfluenten-connect hochgeladen) sollte in etwa so aussehen:

{ 
"name": "cassandra.sink.yourConfigName", 
"config": { 
    "connector.class": "com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraSinkConnector", 
    "tasks.max": "1", 
    "topics": "<your topic>", 
    "connect.cassandra.sink.kcql": "INSERT INTO <your_table> SELECT * FROM <your_kafka_topic>;", 
    "connect.cassandra.contact.points": "<cassandra nodes>", 
    "connect.cassandra.port": "<cassandra port>", 
    "connect.cassandra.key.space": "<cassandra keyspace>", 
    "connect.cassandra.username": "cassandra", 
    "connect.cassandra.password": "cassandra" 
    }}