![\"\"](\"https:\/\/griddb.net\/wp-content\/uploads\/2020\/08\/Screen-Shot-2020-08-07-at-10.35.20-AM.png\")
<\/p>\nIn this tutorial we feed the CASAS Dataset through Kafka and into GridDB with a Kafka Connect Sink that uses GridDB’s new JDBC interface. <\/p>\n
The raw TSV data will be converted into JSON with Gawk and then fed into Kafka with a Console Producer. Then the GridDB JDBC Sink will read the data and write it to GridDB. Finally, we’ll inspect the data with SQLWorkbench\/J.<\/p>\n
<\/p>\n
Kafka<\/a> is a data streaming platform with many different possible inputs and outputs that are easy to create. For this tutorial we’ll use a Kafka Console Producer to put data into Kafka which will be then consumed by Kafka Connect Sink that we are going write. <\/p>\n We are going to follow the Kafka Qucikstart<\/a>. Kafka can be downloaded from their downloads page<\/a>, we’re using version 2.12-2.5.0. You will also need to have a Java 1.8 development environment installed on your system. After downloading, we simply untar and start the Zookeeper and Kafka Servers.<\/p>\n As Zookeeper and Kafka are not daemons by default, have multiple terminals handy that wont terminate (or use screen) if you have network issues. <\/p>\n CASAS<\/a> is research group based out of the Washington State University that has a large sensor dataset<\/a>. The data was collected from a variety of in-home sensors while volunteers performed their normal daily routines. We are going to use this dataset in a series of tutorials to demonstrate how to use GridDB to ingest, analyze, and visualize data. <\/p>\n The data looks like this:<\/p>\n The fields of the TSV file are:<\/p>\n To put the CASAS data into Kafka, we’ll use a Gawk<\/a> (a great text processing tool) to convert the TSV data to JSON that will be then fed into the Kafka Console Producer. If the sensor data was live, it would likely be fed into Kafka via MQTT or an HTTP endpoint. <\/p>\n Here is the simple shell script (casas_convert.sh) takes the CASAS location ID as an argument and uses Gawk to convert the CSV to JSON which is output on stdout looks like this: <\/p>\n Then we can ingest it with Kafka Console Producer: <\/p>\n Two files make up the GridDB JDBC Sink, the first is GridDBSinkConnector.java which primarily handles reading the configuration file and creating the properties that will be used by the second file, GridDBSinkTask.java.<\/p>\n GridDBSinkConnector.java:<\/p>\n GridDBSinkTask.java is more interesting, first the start() method connects to JDBC from the configuration properties that was created by GridDBSinkConnector<\/p>\n The put() method actually writes all data in the subscribed Kafka topic to GridDB using a JDBC prepared statement. One important thing to note is that your prepared statement must be closed in the exception handler, otherwise future prepared statements wont execute. <\/p>\n We use Gradle to build the Jar:<\/p>\n We will assume that you already have GridDB 4.5 CE installed and running the GridDB JDBC Jar built. We will copy GridDB JDBC jar and the GridDB Sink Jar to the Kafka Libs folder. <\/p>\n A configuration file is required for the Sink which defines the Sink Class, the topics it subscribes too as well as any configuration properties that the Sink itself requires. <\/p>\n You will also need to edit configs\/connect-standalone.properties, changing key.converter.schemas.enable and Now we can start the Sink and data will start being ingested if the above console-producer has been run. <\/p>\n\n$ tar xzvf kafka_2.12-2.5.0.tgz\n$ cd kafka_2.12-2.5.0\n$ bin\/zookeeper-server-start.sh config\/zookeeper.properties\n$ bin\/kafka-server-start.sh config\/server.properties\n<\/pre>\n
Data Set<\/h1>\n
\n2012-07-18 12:54:45.126257\tD001\tIgnore\t\tIgnore\t\tCLOSE\tControl4-Door\n2012-07-18 12:54:45.196564\tD002\tOutsideDoor\tFrontDoor\tOPEN\tControl4-Door\n2012-07-18 12:54:45.247825\tT102\tIgnore\t\tFrontDoorTemp\t78\tControl4-Temperature\n2012-07-18 12:54:45.302398\tBATP102\tIgnore\t\tIgnore\t\t85\tControl4-BatteryPercent\n2012-07-18 12:54:45.399416\tT103\tIgnore\t\tBathroomTemp\t25\tControl4-Temperature\n2012-07-18 12:54:45.472391\tBATP103\tIgnore\t\tIgnore\t\t82\tControl4-BatteryPercent\n2012-07-18 12:54:45.606580\tT101\tIgnore\t\tIgnore\t\t31\tControl4-Temperature\n2012-07-18 12:54:45.682577\tMA016\tKitchen\t\tKitchen\t\tOFF\tControl4-MotionArea\n2012-07-18 12:54:45.723461\tD003\tBathroom\tBathroomDoor\tOPEN\tControl4-Door\n2012-07-18 12:54:45.767498\tM009\tBedroom\t\tBedroom\t\tON\tControl4-Motion\n<\/pre>\n
\n#!\/bin\/bash\n\ncat ${1}\/${1}.rawdata.txt | gawk '{ printf \"{\"datetime\" : \"%s %s\", \"sensor\" : \"%s\", \"translate01\": \"%s\", \"translate02\": \"%s\", \"message\":\"%s\", \"sensoractivity\": \"%s\", \"location\": \"'$1'\" }n\", $1, $2, $3, $4, $5, $6, $7 }'\n\n<\/pre>\n\nfor loc in `ls`; do\n .\/casas_covert.sh $loc | \/path\/to\/kafka\/bin\/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic casas\ndone\n<\/pre>\n
Coding the GridDB Sink<\/h1>\n
\n public static final String JDBCURL_CONFIG = \"jdbcurl\";\n public static final String USER_CONFIG = \"user\";\n public static final String PASSWORD_CONFIG = \"password\";\n private static final ConfigDef CONFIG_DEF = new ConfigDef()\n .define(JDBCURL_CONFIG, Type.STRING, Importance.HIGH, \"GridDB JDBC URL\")\n .define(USER_CONFIG, Type.STRING, Importance.HIGH, \"GridDB Username\")\n .define(PASSWORD_CONFIG, Type.STRING, Importance.HIGH, \"GridDB Password\");\n\n String jdbcurl, user, password;\n\n @Override\n public void start(Map
\n @Override public void start(Map
\n @Override public void put(Collection
\n$ .\/gradlew jar \n:compileJava \n:processResources \n:classes \n:jar\n\nBUILD SUCCESSFUL\n<\/pre>\n
Running the GridDB Sink<\/h1>\n
\n$ cp \/path\/to\/griddb-jdbc-sink\/build\/libs\/griddb-jdbc-sink.jar \/path\/to\/kafka\/libs\n$ cp \/path\/to\/griddb-jdbc\/bin\/griddb-jdbc.jar \/path\/to\/kafka\/libs\n<\/pre>\n
\nname=griddb-sink\nconnector.class=net.griddb.connect.griddb.GridDBSinkConnector\ntasks.max=1\ntopics=casas\nschemas.enable=true\njdbcurl=jdbc:gs:\/\/239.0.0.1:41999\/defaultCluster\/public\nuser=admin\npassword=admin\ndatabase=public\n<\/pre>\n
\nvalue.converter.schemas.enable to false.<\/p>\n\n$ cd path\/to\/kafka\n$ .\/bin\/connect-standalone.sh config\/connect-standalone.properties config\/griddb-sink.properties\n[2020-08-07 17:02:38,802] INFO Kafka Connect standalone worker initializing ... (org.apache.kafka.connect.cli.ConnectStandalone:69)\n[2020-08-07 17:53:02,117] INFO Kafka version: 2.5.0 (org.apache.kafka.common.utils.AppInfoParser:117)\n[2020-08-07 17:53:02,117] INFO Kafka commitId: 66563e712b0b9f84 (org.apache.kafka.common.utils.AppInfoParser:118)\n[2020-08-07 17:53:02,117] INFO Kafka startTimeMs: 1596819182117 (org.apache.kafka.common.utils.AppInfoParser:119)\n[2020-08-07 17:53:02,124] INFO Created connector griddb-sink (org.apache.kafka.connect.cli.ConnectStandalone:112)\n[2020-08-07 17:53:02,125] INFO [Consumer clientId=connector-consumer-griddb-sink-0, groupId=connect-griddb-sink] Subscribed to topic(s): casas (org.apache.kafka.clients.consumer.KafkaConsumer:974)\nStarting Task....\n[2020-08-07 17:53:05,901] INFO [Consumer clientId=connector-consumer-griddb-sink-0, groupId=connect-griddb-sink] Adding newly assigned partitions: casas-0 (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:273)\n[2020-08-07 17:53:05,914] INFO [Consumer clientId=connector-consumer-griddb-sink-0, groupId=connect-griddb-sink] Setting offset for partition casas-0 to the committed offset FetchPosition{offset=108465, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[localhost:9092 (id: 0 rack: null)], epoch=0}} (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:792)\nProcessing 500 records...\nWrote {datetime=2011-06-29 21:02:50.709770, translate01=Bathroom, sensoractivity=Control4-MotionArea, translate02=Bathroom, sensor=MA013, location=csh102, message=OFF}\nProcessing 500 records...\nWrote {datetime=2011-06-30 06:29:24.054707, translate01=Ignore, sensoractivity=Control4-LightSensor, translate02=Ignore, sensor=LS006, location=csh102, message=9}\nProcessing 500 records...\nWrote {datetime=2011-06-30 08:29:32.667029, translate01=Kitchen, sensoractivity=Control4-Motion, translate02=Kitchen, sensor=M008, location=csh102, message=ON}\nProcessing 500 records...\nWrote {datetime=2011-06-30 08:55:23.796199, translate01=Bathroom, sensoractivity=Control4-MotionArea, translate02=Bathroom, sensor=MA013, location=csh102, message=OFF}\nProcessing 500 records...\n... \n<\/pre>\nInspecting the Data<\/h1>\n
![\"\"](\"https:\/\/griddb.net\/wp-content\/uploads\/2020\/08\/casas_sqlworkbenchj.png\")
<\/p>\n