External Integrations

Ingesting Data from Kafka

Citus can leverage existing Postgres data ingestion tools. For instance, we can use a tool called kafka-sink-pg-json to copy JSON messages from a Kafka topic into a database table. As a demonstration, we’ll create a kafka_test table and ingest data from the test topic with a custom mapping of JSON keys to table columns.

The easiest way to experiment with Kafka is using the Confluent platform, which includes Kafka, Zookeeper, and associated tools whose versions are verified to work together.

  1. # we're using Confluent 2.0 for kafka-sink-pg-json support
  2. curl -L http://packages.confluent.io/archive/2.0/confluent-2.0.0-2.11.7.tar.gz \
  3.   | tar zx
  5. # Now get the jar and conf files for kafka-sink-pg-json
  6. mkdir sink
  7. curl -L https://github.com/justonedb/kafka-sink-pg-json/releases/download/v1.0.2/justone-jafka-sink-pg-json-1.0.zip -o sink.zip
  8. unzip -d sink $_ && rm $_

The download of kafka-sink-pg-json contains some configuration files. We want to connect to the coordinator Citus node, so we must edit the configuration file sink/justone-kafka-sink-pg-json-connector.properties:

  1. # add to sink/justone-kafka-sink-pg-json-connector.properties
  3. # the kafka topic we will use
  4. topics=test
  6. # db connection info
  7. # use your own settings here
  8. db.host=localhost:5432
  9. db.database=postgres
  10. db.username=postgres
  11. db.password=bar
  13. # the schema and table we will use
  14. db.schema=public
  15. db.table=kafka_test
  17. # the JSON keys, and columns to store them
  18. db.json.parse=/@a,/@b
  19. db.columns=a,b

Notice db.columns and db.json.parse. The elements of these lists match up, with the items in db.json.parse specifying where to find values inside incoming JSON objects.

Note

The paths in db.json.parse are written in a language that allows some flexibility in getting values out of JSON. Given the following JSON,

  1. {
  2.   "identity":71293145,
  3.   "location": {
  4.     "latitude":51.5009449,
  5.     "longitude":-2.4773414
  6.   },
  7.   "acceleration":[0.01,0.0,0.0]
  8. }

here are some example paths and what they match:

  • /@identity - the path to element 71293145.

  • /@location/@longitude - the path to element -2.4773414.

  • /@acceleration/#0 - the path to element 0.01

  • /@location - the path to element {"latitude":51.5009449, "longitude":-2.4773414}

Our own scenario is simple. Our events will be objects like {"a":1, "b":2}. The parser will pull those values into eponymous columns.

Now that the configuration file is set up, it’s time to prepare the database. Connect to the coordinator node with psql and run this:

  1. -- create metadata tables for kafka-sink-pg-json
  2. \i sink/install-justone-kafka-sink-pg-1.0.sql
  4. -- create and distribute target ingestion table
  5. create table kafka_test ( a int, b int );
  6. select create_distributed_table('kafka_test', 'a');

Start the Kafka machinery:

  1. # save some typing
  2. export C=confluent-2.0.0
  4. # start zookeeper
  5. $C/bin/zookeeper-server-start \
  6.   $C/etc/kafka/zookeeper.properties
  8. # start kafka server
  9. $C/bin/kafka-server-start \
  10.   $C/etc/kafka/server.properties
  12. # create the topic we'll be reading/writing
  13. $C/bin/kafka-topics --create --zookeeper localhost:2181   \
  14.                     --replication-factor 1 --partitions 1 \
  15.                     --topic test

Run the ingestion program:

  1. # the jar files for this are in "sink"
  2. export CLASSPATH=$PWD/sink/*
  4. # Watch for new events in topic and insert them
  5. $C/bin/connect-standalone \
  6.   sink/justone-kafka-sink-pg-json-standalone.properties \
  7.   sink/justone-kafka-sink-pg-json-connector.properties

At this point Kafka-Connect is watching the test topic, and will parse events there and insert them into kafka_test. Let’s send an event from the command line.

  1. echo '{"a":42,"b":12}' | \
  2.   $C/bin/kafka-console-producer --broker-list localhost:9092 --topic test

After a small delay the new row will show up in the database.

  1. select * from kafka_test;
  3. ┌────┬────┐
  4. │ a  │ b  │
  5. ├────┼────┤
  6. │ 42 │ 12 │
  7. └────┴────┘

Caveats

  • At the time of this writing, kafka-sink-pg-json requires Kafka version 0.9 or earlier.

  • The kafka-sink-pg-json connector config file does not provide a way to connect with SSL support, so this tool will not work with Citus Cloud which requires secure connections.

  • A malformed JSON string in the Kafka topic will cause the tool to become stuck. Manual intervention in the topic is required to process more events.

Ingesting Data from Spark

People sometimes use Spark to transform Kafka data, such as by adding computed values. In this section we’ll see how to ingest Spark dataframes into a distributed Citus table.

First let’s start a local Spark cluster. It has several moving parts, so the easiest way is to run the pieces with docker-compose.

  1. wget https://raw.githubusercontent.com/gettyimages/docker-spark/master/docker-compose.yml
  3. # this may require "sudo" depending on the docker daemon configuration
  4. docker-compose up

To do the ingestion into PostgreSQL, we’ll be writing custom Scala code. We’ll use the Scala Build Tool (SBT) to load dependencies and run our code, so download SBT and install it on your machine.

Next create a new directory for our project.

  1. mkdir sparkcitus

Create a file called sparkcitus/build.sbt to tell SBT our project configuration, and add this:

  1. // add this to build.sbt
  3. name := "sparkcitus"
  4. version := "1.0"
  6. scalaVersion := "2.10.4"
  8. resolvers ++= Seq(
  9.   "Maven Central" at "http://central.maven.org/maven2/"
  10. )
  12. libraryDependencies ++= Seq(
  13.   "org.apache.spark" %% "spark-core" % "2.2.1",
  14.   "org.apache.spark" %% "spark-sql"  % "2.2.1",
  15.   "org.postgresql"   %  "postgresql" % "42.2.2"
  16. )

Next create a helper Scala class for doing ingestion through JDBC. Add the following to sparkcitus/copy.scala:

  1. import java.io.InputStream
  2. import java.sql.DriverManager
  3. import java.util.Properties
  5. import org.apache.spark.sql.{DataFrame, Row}
  6. import org.postgresql.copy.CopyManager
  7. import org.postgresql.core.BaseConnection
  9. object CopyHelper {
  11.   def rowsToInputStream(rows: Iterator[Row]): InputStream = {
  12.     val bytes: Iterator[Byte] = rows.map { row =>
  13.       (row.toSeq
  14.         .map { v =>
  15.           if (v == null) {
  16.             """\N"""
  17.           } else {
  18.             "\"" + v.toString.replaceAll("\"", "\"\"") + "\""
  19.           }
  20.         }
  21.         .mkString("\t") + "\n").getBytes
  22.     }.flatten
  24.     new InputStream {
  25.       override def read(): Int =
  26.         if (bytes.hasNext) {
  27.           bytes.next & 0xff // make the signed byte an unsigned int
  28.         } else {
  29.           -1
  30.         }
  31.     }
  32.   }
  34.   def copyIn(url: String, df: DataFrame, table: String):Unit = {
  35.     var cols = df.columns.mkString(",")
  37.     df.foreachPartition { rows =>
  38.       val conn = DriverManager.getConnection(url)
  39.       try {
  40.         val cm = new CopyManager(conn.asInstanceOf[BaseConnection])
  41.         cm.copyIn(
  42.           s"COPY $table ($cols) " + """FROM STDIN WITH (NULL '\N', FORMAT CSV, DELIMITER E'\t')""",
  43.           rowsToInputStream(rows))
  44.         ()
  45.       } finally {
  46.         conn.close()
  47.       }
  48.     }
  49.   }
  50. }

Continuing the setup, save some sample data into people.json. Note the intentional lack of surrounding square brackets. Later we’ll create a Spark dataframe from the data.

  1. {"name":"Tanya Rosenau"   , "age": 24},
  2. {"name":"Rocky Slay"      , "age": 85},
  3. {"name":"Tama Erdmann"    , "age": 48},
  4. {"name":"Jared Olivero"   , "age": 42},
  5. {"name":"Gudrun Shannon"  , "age": 53},
  6. {"name":"Quentin Yoon"    , "age": 32},
  7. {"name":"Yanira Huckstep" , "age": 53},
  8. {"name":"Brendon Wesley"  , "age": 19},
  9. {"name":"Minda Nordeen"   , "age": 79},
  10. {"name":"Katina Woodell"  , "age": 83},
  11. {"name":"Nevada Mckinnon" , "age": 65},
  12. {"name":"Georgine Mcbee"  , "age": 56},
  13. {"name":"Mittie Vanetten" , "age": 17},
  14. {"name":"Lecia Boyett"    , "age": 37},
  15. {"name":"Tobias Mickel"   , "age": 69},
  16. {"name":"Jina Mccook"     , "age": 82},
  17. {"name":"Cassidy Turrell" , "age": 37},
  18. {"name":"Cherly Skalski"  , "age": 29},
  19. {"name":"Reita Bey"       , "age": 69},
  20. {"name":"Keely Symes"     , "age": 34}

Finally, create and distribute a table in Citus:

  1. create table spark_test ( name text, age integer );
  2. select create_distributed_table('spark_test', 'name');

Now we’re ready to hook everything together. Start up sbt:

  1. # run this in the sparkcitus directory
  3. sbt

Once inside sbt, compile the project and then go into the “console” which is a Scala repl that loads our code and dependencies:

  1. sbt:sparkcitus> compile
  2. [success] Total time: 3 s
  4. sbt:sparkcitus> console
  5. [info] Starting scala interpreter...
  7. scala>

Type these Scala commands into the console:

  1. // inside the sbt scala interpreter
  3. import org.apache.spark.sql.SparkSession
  5. // open a session to the Spark cluster
  6. val spark = SparkSession.builder().appName("sparkcitus").config("spark.master", "local").getOrCreate()
  8. // load our sample data into Spark
  9. val df = spark.read.json("people.json")
  11. // this is a simple connection url (it assumes Citus
  12. // is running on localhost:5432), but more complicated
  13. // JDBC urls differ subtly from Postgres urls, see:
  14. // https://jdbc.postgresql.org/documentation/head/connect.html
  15. val url = "jdbc:postgresql://localhost/postgres"
  17. // ingest the data frame using our CopyHelper class
  18. CopyHelper.copyIn(url, df, "spark_test")

This uses the CopyHelper to ingest the ionformation. At this point the data will appear in the distributed table.

Note

Our method of ingesting the dataframe is straightforward but doesn’t protect against Spark errors. Spark guarantees “at least once” semantics, i.e. a read error can cause a subsequent read to encounter previously seen data.

A more complicated, but robust, approach is to use the custom Spark partitioner spark-citus so that partitions match up exactly with Citus shards. This allows running transactions directly on worker nodes which can rollback on read failure. See the presentation linked in that repository for more information.

Business Intelligence with Tableau

Tableau is a popular business intelligence and analytics tool for databases. Citus and Tableau provide a seamless experience for performing ad-hoc reporting or analysis.

You can now interact with Tableau using the following steps.

  • Choose PostgreSQL from the “Add a Connection” menu.

    postgres option selected in menu

  • Enter the connection details for the coordinator node of your Citus cluster. (Note if you’re connecting to Citus Cloud you must select “Require SSL.”)

    postgres connection details form

  • Once you connect to Tableau, you will see the tables in your database. You can define your data source by dragging and dropping tables from the “Table” pane. Or, you can run a custom query through “New Custom SQL”.

  • You can create your own sheets by dragging and dropping dimensions, measures, and filters. You can also create an interactive user interface with Tableau. To do this, Tableau automatically chooses a date range over the data. Citus can compute aggregations over this range in human real-time.

united states map with bar chart