Cassandra CDC — Lenses

NoteThis is still beta versionKafka Connect Cassandra is a Source Connector for reading Change Data Capture from Cassandra and write the mutations to Kafka.Why¶We already provide a Kafka Connect Cassandra Source and people might ask us why another source. The main reason is performance. We have noticed our users with large data in a Cassandra table (column family)the performance drops. And this is not on the connector but rather how long it takes to pull the records. We have worked around the issue by introducing a limit on the records queried on Cassandra but evenso, we are not that much better.With Apache Cassandra 3.0 the support for CDC(Change Data Capture) has been added and this source implementation is making use of that to capture the changes made to your column families.Cassandra CDC¶Change data capture is designed to capture insert, update, and delete activity applied to tables(column families), and to make the details of the changes available in an easily consumed format.Cassandra CDC logging is configured per table, with limits on the amount of disk space to consume for storing the CDC logs. CDC logs use the same binary format as the commit log.Cassandra tables can be created or altered with a table property to use CDC logging.CREATE TABLE foo (a int, b text, PRIMARY KEY(a)) WITH cdc=true;ALTER TABLE foo WITH cdc=true;ALTER TABLE foo WITH cdc=false;CDC logging must be enabled in the cassandra.yaml file to begin logging. You should make sure your yaml file has the following:cdc_enabled: trueYou can enable the CDC logging on per node basis.The Kafka Connect Source consumes the CDC log information and pushes it to Kafka before it deletes it.WarningUpon flushing the memtable to disk, CommitLogSegments containing data for CDC-enabled tables are moved to theconfigured cdc_raw directory. Once the disk space limit is reached, writes to CDC enabled tables will be rejecteduntil space is freed.Four CDC settings are configured in the cassandra.yamlcdc_enabledEnables/Disables CDC logging per nodecdc_raw_directoryThe directory where the CDC log is stored.Package installations(default): $CASSANDRA_HOME/cdc_raw.Tarball installations: install_location/data/cdc_raw.cdc_total_space_in_mbTotal space available for storing CDC data. The default is 4096MB and 1/8th of the total space of the drive where thecdc_raw_directory resides. If space gets above this value, Cassandra will throw WriteTimeoutException on Mutationsincluding tables with CDC enabled.cdc_free_space_check_interval_msWhen the cdc_raw limit is hit and the Consumer is either running behind or experiencing back pressure, this interval ischecked to see if any new space for cdc-tracked tables has been made available.WarningAfter changing properties in the cassandra.yaml file, you must restart the node for the changes to take effect.Prerequisites¶Apache Kafka 0.11.x or aboveKafka Connect 0.11.x or aboveCassandra 3.0.9+ or aboveJava 1.8Cassandra Setup¶First, download and install Cassandra if you don’t have a compatiblecluster available.#make a folder for cassandramkdir cassandra#Download Cassandrawget http://apache.mirror.anlx.net/cassandra/3.11.0/apache-cassandra-3.11.0-bin.tar.gz#extract archive to cassandra foldertar xvf apache-cassandra-3.11.0-bin.tar.gz -C cassandra --strip-components=1#enable the CDC in the yaml configurationsed -i -- 's/cdc_enabled: false/cdc_enabled: true/g' conf/cassandra.yaml#set CASSANDRA_HOMEexport CASSANDRA_HOME=$(pwd)/cassandra#Start Cassandracd cassandrasudo sh /bin/cassandraNoteThere can be only one instance of Apache Cassandra node per machine for the connector to run properly. All nodesshould have the same path for the cdc_raw folderSource Connector¶The Cassandra CDC Source connector will read the commit log mutations from the CDC logs and will push them to the target topic.Notemessages sent to Kafka are in AVRO format.The record pushed to Kafka populates both the key and the value part. Key will contain the metadata of the change whilethe value will contain the actual data change.Record Key¶The key data structure follows this layout:{ "keyspace": //Cassandra Keyspace name "table" : //The Cassandra Column Family name "changeType": //The type of change in Cassandra "keys": { "key1": "key2": .. } "timestamp" : //the timestamp of when the change was made in Cassandra "deleted_columns": //which columns have been deleted. We will expand on the details}Based on the mutation information we can identify the following types of changes:INSERTA record has been inserted/a record columns have been updated. There is no real solution for identifying anUPDATE unless the connector keeps track of all the keys seen.INSERT INTO keyspace.orders (id, created, product, qty, price) VALUES (1, now(), 'OP-DAX-P-20150201-95.7', 100, 94.2)DELETEAn entire record has been deleted (tombstoned)DELETE FROM datamountaineer.orders where id = 1DELETE_COLUMNSpecific columns have been deleted (non PK columns).DELETE product FROM datamountaineer.orders where id = 1DELETE name.firstname FROM datamountaineer.users WHERE id=62c36092-82a1-3a00-93d1-46196ee77204;In this case the deleted_columns entry will contain “product/name.firstname”. If more than one column is deleted wewill retain that information.Value Key¶The Kafka message value part contains the actual mutation data. Apart from the primary keys columns, all the other columnshave an optional schema in AVRO. The reason for that is because one can set the values on a subset of them during aKCQL insert/update. In the QuickStart section, we make use of the users table. The value AVRO schema associated withit looks like this{ "type" : "record", "name" : "users", "fields" : [ { "name" : "name", "type" : [ "null", { "type" : "record", "name" : "fullname", "fields" : [ { "name" : "firstname", "type" : [ "null", "string" ], "default" : null }, { "name" : "lastname", "type" : [ "null", "string" ], "default" : null } ], "connect.name" : "fullname" } ], "default" : null }, { "name" : "addresses", "type" : [ "null", { "type" : "array", "items" : { "type" : "record", "name" : "MapEntry", "namespace" : "io.confluent.connect.avro", "fields" : [ { "name" : "key", "type" : [ "null", "string" ], "default" : null }, { "name" : "value", "type" : [ "null", { "type" : "record", "name" : "address", "namespace" : "", "fields" : [ { "name" : "street", "type" : [ "null", "string" ], "default" : null }, { "name" : "city", "type" : [ "null", "string" ], "default" : null }, { "name" : "zip_code", "type" : [ "null", "int" ], "default" : null }, { "name" : "phones", "type" : [ "null", { "type" : "array", "items" : [ "null", "string" ] } ], "default" : null } ], "connect.name" : "address" } ], "default" : null } ] } } ], "default" : null }, { "name" : "direct_reports", "type" : [ "null", { "type" : "array", "items" : [ "null", "fullname" ] } ], "default" : null }, { "name" : "id", "type" : [ "null", "string" ], "default" : null }, { "name" : "other_reports", "type" : [ "null", { "type" : "array", "items" : [ "null", "fullname" ] } ], "default" : null } ], "connect.name" : "users"}And a json representation of the actual value looks like this:{ "id" : "UUID-String", "name": { "firstname":"String" "lastname":"String" }, "direct_reports":[ { "firstname":"String" "lastname":"String" }, ... ], "other_reports":[ { "firstname":"String" "lastname":"String" }, ... ], "addresses" : { "home" : { "street": "String", "city": "String", "zip_code": "int", "phones": [ "+33 ...", "+33 ..." ] }, "work" : { "street": "String", "city": "String", "zip_code": "int", "phones": [ "+33 ...", "+33 ..." ] }, ... }}Data Types¶The Source connector needs to map Apache Cassandra types to Kafka Connect Schema types. For the ones not so familiarwith connect here is the list of supported Connect Types:INT8INT16INT32INT64FLOAT32FLOAT64BOOLEANSTRINGBYTESARRAYMAPSTRUCTAlong these primitive types, there are the logical types of:DateDecimalTimeTimestampAs a result, for most Apache Cassandra Types, we have an equivalent type for Connect Schema. A Connect Source Record willbe marshaled as AVRO when sent to Kafka.CQL TypeConnect Data TypeAsciiTypeOPTIONAL STRINGLongTypeOPTIONAL INT64BytesTypeOPTIONAL BYTESBooleanTypeOPTIONAL BOOLEANCounterColumnTypeOPTIONAL INT64SimpleDateTypeOPTIONAL Kafka Connect DateDoubleTypeOPTIONAL FLOAT64DecimalTypeOPTIONAL Kafka Connect DecimalDurationTypeOPTIONAL STRINGEmptyTypeOPTIONAL STRINGFloatTypeOPTIONAL FLOAT32InetAddressTypeOPTIONAL STRINGInt32TypeOPTIONAL INT32ShortTypeOPTIONAL INT16UTF8TypeOPTIONAL STRINGTimeTypeOPTIONAL KAFKA CONNECT TimeTimestampTypeOPTIONAL KAFKA CONNECT TimestampTimeUUIDTypeOPTIONAL STRINGByteTypeOPTIONAL INT8UUIDTypeOPTIONAL STRINGIntegerTypeOPTIONAL INT32ListTypeOPTIONAL ARRAY of the inner typeMapTypeOPTIONAL MAP of the inner typesSetTypeOPTIONAL ARRAY of the inner typeUserTypeOPTIONAL STRUCT for the user typePlease note we default to String for the these CQL types: DurationType, InetAddressType, TimeUUIDType, UUIDType.How does it work¶It is expected that Kafka Connect worker will run on the same node as the Apache Cassandra node.ImportantOnly one Apache Cassandra Node should run per machine to have the Connector work properlyThe cdc_raw folder location should be the same on all nodes running Apache Cassandra NodeThere should be only one Connector Worker instance per machine. Any more won’t have any effectCassandra supports a master-less “ring” architecture. Each of the nodes in the Cassandra ring cluster will beresponsible for storing the table records. The partition key hash and number of rings in the cluster determines the whereeach record is stored. (we leave aside replication from this discussion).Upon flushing the memtable to disk, all the commit log segments containing data for CDC-enabled tables are moved to theconfigured cdc_raw directory. It is only at this point the connector will pick up the changes.ImportantChanges in Cassandra are not picked up immediately. The memtables need to be flushed for the CDC commit logs to be available.You can use nodetool to flush the tablesOnce a file lands in the CDC folder the Connector will pick it up and read the mutations. A CDC file can contain mutationsfor more than one table. Each mutation for the subscribed tables will be translated into a Kafka Connect Source Recordwhich will be sent by the Connect framework to the topic configured in the connector properties.The Connect source will process the files in the order they were created and one by one. This ensures the changesequence is retained. Once the records have been pushed to Kafka the CDC file is deleted.The connector will only be able to read the mutations for the subscribed tables. Via configuration you can express whichtables to consider and what topic should receive those mutations information.INSERT INTO ordersTopic SELECT * FROM datamountaineer.ordersImportantEnabling CDC on a new table means you need to restart the connector for the changes to be picked up. The connector isdriven by the configurations and not by the list of all the tables with CDC enabled. (Might be a feature, change to do)Below you can find a flow diagram describing the process mentioned above.Source Connector QuickStart¶We will start the connector in distributed mode. Each connector exposes a rest endpoint for stopping, starting andupdating the configuration. We have developed a Command Line Interface to make interacting with the Connect Rest APIeasier. The CLI can be found in the Stream Reactor download under the bin folder. Alternatively the Jar can bepulled from our GitHub releases page.Once you have installed and started Cassandra create a table to capture the mutations. We will use a bit more complexcolumn family structure to showcase what we support so far.Let’s start the cql shell tool to create our table$./bin/cqlshConnected to Test Cluster at 127.0.0.1:9042. [cqlsh 5.0.1 | Cassandra 3.11.0 | CQL spec 3.4.4 | Native protocol v4] Use HELP for help. cqlsh>Now let’s create a keyspace and a users column family(table)CREATE KEYSPACE datamountaineer WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 3};CREATE TYPE datamountaineer.address( street text, city text, zip_code int, phones set<text>);CREATE TYPE datamountaineer.fullname ( firstname text, lastname text);CREATE TABLE datamountaineer.users( id uuid PRIMARY KEY, name fullname, direct_reports set<frozen <fullname>>, other_reports list<frozen <fullname>>, addresses map<text, frozen <address>>);ALTER TABLE datamountaineer.users WITH cdc=true;Starting the Connector (Distributed)¶Download, and install Stream Reactor to your Kafka Connect cluster. Follow the instructions here if you haven’t already done so.All paths in the quickstart are based in the location you installed Stream Reactor.Start Kafka Connect in distributed more by running:#make sure you have $CASSANDRA_HOME variable setup (see Cassandra setup)bin/connect-distributed.sh etc/schema-registry/connect-avro-distributed.properties#make sure you have $CASSANDRA_HOME variable setup (see Cassandra setup)$ cat <<EOF > cassandra-cdc-source.json{ "name": "cassandra-connect-cdc", "config":{ "name":"cassandra-connct-cdc", "tasks": 1, "connector.class":"com.datamountaineer.streamreactor.connect.cassandra.cdc.CassandraCdcSourceConnector", "connect.cassandra.kcql":"INSERT INTO users-topic SELECT * FROM datamountaineer.users", "connect.cassandra.yaml.path.url": "$CASSANDRA_HOME/conf/cassandra.yaml", "connect.cassandra.port": "9042", "connect.cassandra.contact.points":"localhost" }}EOFIf you visualize the file it should print something like (I used the .. because you might have a different path to where you stored Cassandra){ "name": "cassandra-connect-cdc", "config":{ "name":"cassandra-connct-cdc", "tasks": 1, "connector.class":"com.datamountaineer.streamreactor.connect.cassandra.cdc.CassandraCdcSourceConnector", "connect.cassandra.kcql":"INSERT INTO orders-topic SELECT * FROM datamountaineer.orders", "connect.cassandra.yaml.path.url": "../cassandra/conf/cassandra.yaml", "connect.cassandra.port": "9042", "connect.cassandra.contact.points":"localhost" }}Next step is to spin up the connector. And for that we run the following bash script:curl -X POST -H "Content-Type: application/json" --data @cassandra-cdc-source.json http://localhost:8083/connectorsThe output from the connect-distributed should read something similar to this: [2017-08-01 22:34:35,653] INFO Acquiring port 64101 to enforce single instance being run on Stepi (com.datamountaineer.streamreactor.connect.cassandra.cdc.CassandraCdcSourceTask:53) [2017-08-01 22:34:35,660] INFO __ __ / / ____ _____ ____/ /___ ____ ____ / / / __ `/ __ \/ __ / __ \/ __ \/ __ \ / /___/ /_/ / / / / /_/ / /_/ / /_/ / /_/ //_____/\__,_/_/ /_/\__,_/\____/\____/ .___/ /_/ ____ by Stefan Bocutiu _ ____ ____ ____ / ___|__ _ ___ ___ __ _ _ __ __| |_ __ __ _ / ___| _ \ / ___| | | / _` / __/ __|/ _` | '_ \ / _` | '__/ _` | | | | | | | | | |__| (_| \__ \__ \ (_| | | | | (_| | | | (_| | | |___| |_| | |___ \____\__,_|___/___/\__,_|_| |_|\__,_|_| \__,_| \____|____/ \____| (com.datamountaineer.streamreactor.connect.cassandra.cdc.CassandraCdcSourceTask:65) [2017-08-01 22:34:36,088] INFO CDC path is not set in Yaml. Using the default location (com.datamountaineer.streamreactor.connect.cassandra.cdc.logs.CdcCassandra:55) [2017-08-01 22:34:36,411] INFO Detected Guava >= 19 in the classpath, using modern compatibility layer (com.datastax.driver.core.GuavaCompatibility:132)Let’s go back to the cqlsh terminal and insert some records into the users’ table and then perform some updates and deletes.INSERT INTO datamountaineer.users(id, name) VALUES (62c36092-82a1-3a00-93d1-46196ee77204,{firstname:'Marie-Claude',lastname:'Josset'});UPDATE datamountaineer.users SET addresses = addresses + { 'home': { street: '191 Rue St. Charles', city: 'Paris', zip_code: 75015, phones: {'33 6 78 90 12 34'} }, 'work': { street: '81 Rue de Paradis', city: 'Paris', zip_code: 7500, phones: {'33 7 12 99 11 00'} }}WHERE id=62c36092-82a1-3a00-93d1-46196ee77204;INSERT INTO datamountaineer.users(id, direct_reports) VALUES (11c11111-82a1-3a00-93d1-46196ee77204,{{firstname:'Jean-Claude',lastname:'Van Damme'}, {firstname:'Arnold', lastname:'Schwarzenegger'}});INSERT INTO datamountaineer.users(id, other_reports) VALUES (22c11111-82a1-3a00-93d1-46196ee77204,[{firstname:'Jean-Claude',lastname:'Van Damme'}, {firstname:'Arnold', lastname:'Schwarzenegger'}]);DELETE name.firstname FROm datamountaineer.users WHERE id=62c36092-82a1-3a00-93d1-46196ee77204;You will notice from the logs there are no new CDC files picked up and if you navigate to the CDC output folder you will see it is empty. The memtables needs to fill up to be flushed to disk.Let’s use the tool provided by Apache Cassandra to flush the table:nodetool$ $CASSANDRA_HOME/bin/nodetool drainOnce this completes your connect distributed log should print something along these lines:[2017-08-01 23:10:42,842] INFO Reading mutations from the CDC file:/home/stepi/work/programs/cassandra/data/cdc_raw/CommitLog-6-1501625205002.log. Checking file is still being written... (com.datamountaineer.streamreactor.connect.cassandra.cdc.logs.CdcCassandra:158)[2017-08-01 23:10:43,352] INFO Global buffer pool is enabled, when pool is exhausted (max is 0.000KiB) it will allocate on heap (org.apache.cassandra.utils.memory.BufferPool:230)[2017-08-01 23:10:43,355] INFO Maximum memory usage reached (0.000KiB), cannot allocate chunk of 1.000MiB (org.apache.cassandra.utils.memory.BufferPool:91)[2017-08-01 23:10:43,390] ERROR [Control connection] Cannot connect to any host, scheduling retry in 4000 milliseconds (com.datastax.driver.core.ControlConnection:153)[2017-08-01 23:10:43,569] INFO 5 changes detected in /home/stepi/work/programs/cassandra/data/cdc_raw/CommitLog-6-1501625205002.log (com.datamountaineer.streamreactor.connect.cassandra.cdc.logs.CdcCassandra:173)Let’s see what was sent over to the users topic. We will run kafka-avro-console-consumer to read the records$ ./bin/kafka-avro-console-consumer --zookeeper localhost:2181 --topic users-topic --from-beginning --property print.key=trueUsing the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].{"keyspace":"datamountaineer","table":"users","changeType":"INSERT","deleted_columns":null,"keys":{"id":{"string":"62c36092-82a1-3a00-93d1-46196ee77204"}},"timestamp":1501625394958965} {"name":{"fullname":{"firstname":{"string":"Marie-Claude"},"lastname":{"string":"Josset"}}},"addresses":null,"direct_reports":null,"id":{"string":"62c36092-82a1-3a00-93d1-46196ee77204"},"other_reports":null}{"keyspace":"datamountaineer","table":"users","changeType":"INSERT","deleted_columns":null,"keys":{"id":{"string":"62c36092-82a1-3a00-93d1-46196ee77204"}},"timestamp":1501625395008930} {"name":null,"addresses":{"array":[{"key":{"string":"work"},"value":{"address":{"street":{"string":"81 Rue de Paradis"},"city":{"string":"Paris"},"zip_code":{"int":7500},"phones":{"array":[{"string":"33 7 12 99 11 00"}]}}}},{"key":{"string":"home"},"value":{"address":{"street":{"string":"191 Rue St. Charles"},"city":{"string":"Paris"},"zip_code":{"int":75015},"phones":{"array":[{"string":"33 6 78 90 12 34"}]}}}}]},"direct_reports":null,"id":{"string":"62c36092-82a1-3a00-93d1-46196ee77204"},"other_reports":null}{"keyspace":"datamountaineer","table":"users","changeType":"INSERT","deleted_columns":null,"keys":{"id":{"string":"11c11111-82a1-3a00-93d1-46196ee77204"}},"timestamp":1501625395013654} {"name":null,"addresses":null,"direct_reports":{"array":[{"fullname":{"firstname":{"string":"Arnold"},"lastname":{"string":"Schwarzenegger"}}},{"fullname":{"firstname":{"string":"Jean-Claude"},"lastname":{"string":"Van Damme"}}}]},"id":{"string":"11c11111-82a1-3a00-93d1-46196ee77204"},"other_reports":null}{"keyspace":"datamountaineer","table":"users","changeType":"INSERT","deleted_columns":null,"keys":{"id":{"string":"22c11111-82a1-3a00-93d1-46196ee77204"}},"timestamp":1501625395015668} {"name":null,"addresses":null,"direct_reports":null,"id":{"string":"22c11111-82a1-3a00-93d1-46196ee77204"},"other_reports":{"array":[{"fullname":{"firstname":{"string":"Jean-Claude"},"lastname":{"string":"Van Damme"}}},{"fullname":{"firstname":{"string":"Arnold"},"lastname":{"string":"Schwarzenegger"}}}]}}{"keyspace":"datamountaineer","table":"users","changeType":"DELETE_COLUMN","deleted_columns":{"array":["name.firstname"]},"keys":{"id":{"string":"62c36092-82a1-3a00-93d1-46196ee77204"}},"timestamp":1501625395018481} {"name":null,"addresses":null,"direct_reports":null,"id":{"string":"62c36092-82a1-3a00-93d1-46196ee77204"},"other_reports":null}Exactly what was changed!!!Features¶Kafka Connect Query Language¶Both connectors support K afka C onnect Q uery L anguage found hereGitHub repo allows for routing and mapping usinga SQL like syntax, consolidating typically features into one configuration option.INSERT INTO <topic> SELECT * FROM <KEYSPACE>.<TABLE>#Select all mutations for datamountaineer.ordersINSERT INTO ordersTopic SELECT * FROM datamountaineer.orders#while KSQL allows for fields (column) addressing the CDC Source will not use them. It pushes the entire Cassandra mutation information on the topicConfigurations¶Here is a full list of configuration entries the connector knows about.NameDescriptionDataTypeOptionalDefaultconnect.cassandra.contact.pointsContact points (hosts) instringno connect.cassandra.portCassandra Node Client connection portintyes9042connect.cassandra.usernameUsername to connect to Cassandra with ifconnect.cassandra.authentication.modeis set to username_passwordstringyes connect.cassandra.passwordPassword to connect to Cassandra with ifconnect.cassandra.authentication.modeis set to username_password.stringyes connect.cassandra.ssl.enabledEnables SSL communication against SSL enabledCassandra cluster.booleanyesfalseconnect.cassandra.trust.store.passwordPassword for truststore.stringyes connect.cassandra.key.store.pathPath to truststore.stringyes connect.cassandra.key.store.passwordPassword for key store.stringyes connect.cassandra.ssl.client.cert.authPath to keystore.stringyes connect.cassandra.kcqlKafka connect query language expression.Allows for expressive table to topicrouting. It describes which CDC tablesare monitored and the target Kafka topic forCassandra CDC information.stringno connect.cassandra.cdc.path.urlThe location of the Cassandra Yaml file inURL format:file://path.The connector reads the file to get the cdcfolder but also sets the internals ofthe Cassandra API allowing it to read theCDC filesstringno connect.cassandra.cdc.file.watch.intervalThe delay time in millisecondsbefore the connector checks forCassandra CDC files. We pollthe CDC folder for new files.longyes2000connect.cassandra.cdc.mutation.queue.sizeThe maximum number of Cassandra mutationto buffer. As it reads from the Cassandra CDCfiles the mutations are buffered before theyare handed over to Kafka Connect.intyes1000000connect.cassandra.cdc.enable.delete.while.readThe worker CDC thread will read a CDC filechecking if any of the processed files areready to be deleted (that means the recordshave been sent to kafka). Rather than waitingfor a read to complete we can delete thefiles while reading a CDC file.a CDC file. You can disable it for fasterfaster reads by setting the value to false.booleanyesfalseconnect.cassandra.cdc.single.instance.portKafka Connect framework doesn’t allow yetconfiguration where you are running onlyone task per worker. If you allocate moretasks than workers then some will spin upmore tasks. With Cassandra nodes we want onewant one worker and one task - not more.To ensure this we allow the first task tograb a port subsequent calls to open theport will fail thus not allowing multipleinstance running at onceintyes64101connect.cassandra.cdc.decimal.scaleWhen reading the column family metadata wedon’t have details about decimal scale   Deployment Guidelines¶TODOTroubleShooting¶Please review the FAQs and join our slack channel