Data Integration as Service:  Kafka-Connectの紹介と実験

Apache Kafka 0.9より同梱されているKafka Connectを紹介します。

# Kafka-Connect


Kafka ConnectはKafkaと周辺のシステム間でストリームデータをやりとりするための通信規格とライブラリとツールです。まずは下の図をご覧ください。

![kafka-connect.png](https://qiita-image-store.s3.amazonaws.com/0/13775/ee9228dd-4f8f-6806-fbb5-e6b8123e96cc.png "kafka-connect.png")


コネクタは周辺のシステムからKafkaへデータを取り込むための`ソース`と周辺システムへデータを送る`シンク`の二種類があります。データの流れは一方通行です。すでに何十ものコネクタが実装されており、サポートされている周辺システムは多種に渡ります。もちろん自分でコネクタを作ることもできます。

Kafkaの中を通過するデータの形式は基本的にAvroです。コネクタが周辺システム固有のデータ形式をAvroに変換統一しています。これによりどんなソースとシンクの組み合わせでも動くことになります。

Avroには[Schema Registry](https://github.com/confluentinc/schema-registry)があり、スキーマの変更にも対応できます。

コネクタはスタンドアローンモードと分散モードがあり、分散モードでは複数台のサーバーで分散処理することでスケーラビリティと耐障害性を確保しています。

Kafka ConnectのランタイムにはREST APIが装備されていて、複数のコネクタの起動/停止/ステータスの管理が行えるようになっています。

Kafkaパーティション・オフセットの管理も行なっており、障害から回復した時に、データの重複欠損がないことを保証する仕組みもあります。


https://kafka.apache.org/documentation#connect


# 導入

KafkaはApacheのプロジェクトですが、開発の主導はConfluentです。ここではConfluent PlatformでKafkaを導入します。

環境: Ubuntu14 都合により古いバージョンです...

ここにあるドキュメントに従います。Scalaのバージョンは2.11
http://docs.confluent.io/3.0.0/installation.html#installation-apt

## パッケージインストール

```bash
wget -qO - http://packages.confluent.io/deb/3.0/archive.key | sudo apt-key add -
sudo add-apt-repository "deb [arch=amd64] http://packages.confluent.io/deb/3.0 stable main"
sudo apt-get update && sudo apt-get install confluent-platform-2.11
```

## Oracle Java 1.7をインストール
最新のUbuntuならこのステップは不要でしょう。

```bash
sudo add-apt-repository ppa:webupd8team/java
sudo apt-get update
sudo apt-get install oracle-java7-installer
sudo apt-get install oracle-java7-set-default
```

## confluentユーザーを追加

```bash
sudo adduser confluent
```

## 各種ディレクトリ作成

```bash
mkdir /var/log/kafka
mkdir /var/lib/zookeeper
mkdir /var/lib/kafka
chown -R confluent:confluent /var/log/kafka /var/lib/zookeeper /var/lib/kafka
```

サードパーティーのコネクタのjarはkafka-connect-XXXXという名称で/usr/share/javaに保存します。Kafka ConnectのREST APIで起動の指示をした時、認識して自動的にロードしてくれます。

## Zookeeper, Kafka, Schema-Registry 起動

```bash
su - confluent
/usr/bin/zookeeper-server-start /etc/kafka/zookeeper.properties &
/usr/bin/kafka-server-start /etc/kafka/server.properties &
/usr/bin/schema-registry-start /etc/schema-registry/schema-registry.properties &
```

## Kafka-Connect起動

distributedモードで起動します。

```bash
connect-distributed /etc/schema-registry/connect-avro-distributed.properties
```

REST APIを叩いてみると、コネクターのリストは空。まだ何も走らせていないので当たり前。

```bash
$ curl http://localhost:8083/connectors | jq .
[]
```

# syslogをElasticsearchに保存してみる実験
syslogソースコネクタとelasticsearchシンクコネクタを使ってsyslogの内容をElasticsearchに保存してみます。Elasticsearchは2.x系を別途インストールしてください。

## syslogの設定

Ubuntu14はrsyslogを使っていますので、TCPソケット、ポート5514にログを書き込むようにします

`*.* @@localhost:5514`という行を/etc/rsyslog.d/50-default.confの先頭に書きこみ、rsyslogを再起動します。

```bash
service rsyslog restart
```

## syslogソースコネクタ

syslogソースを次のように起動します。コネクタがポート5514でログを受け取るようになります。

```bash
curl -H "Content-Type: application/json" -X POST http://localhost:8083/connectors -d '
{
    "name": "tcpsyslog",
    "config": {
        "tasks.max":1,
        "connector.class":"io.confluent.kafka.connect.syslog.source.TCPSyslogSourceConnector",
        "kafka.topic":"syslog-tcp",
        "syslog.port":5514,
        "syslog.reverse.dns.remote.ip": true
    }
}
'
```

## Elasticsearchシンクコネクタ

同様にElasticsearchシンクを起動します。

```bash
curl -H "Content-Type: application/json" -X POST http://localhost:8083/connectors -d '
{
    "name": "elasticsearch-sink",
    "config": {
        "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
        "tasks.max": 1,
        "topics": "syslog-tcp",
        "key.ignore": true,
        "connection.url": "http://localhost:9200",
        "type.name": "kafka-connect"
    }
}
'
```

# 動作確認

コネクタ一覧を見てみるとtcpsyslogとelasticsearch-sinkコネクタが見えます。

```bash
$ curl http://localhost:8083/connectors | jq .
[
  "elasticsearch-sink",
  "tcpsyslog"
]
```

しばらくするとsyslogが何か書き込んでいるはずですのでESをチェックして見ましょう。

```bash
$ curl http://localhost:9200/_cat/indices
red open syslog-tcp    5 1
```

Kafkaのトピックと同じsyslog-tcpインデックスが作られていました！
読み出して見ましょう。

```bash
$ curl http://localhost:9200/syslog-tcp/_search?pretty
  ...
  {
    "_source": {
      "hostname": "localhost",
      "remote_address": "localhost/127.0.0.1:51631",
      "charset": "UTF-8",
      "message": "vagrant-ubuntu-trusty-64 kernel: [16583.344413] Hardware name: innotek GmbH VirtualBox/VirtualBox, BIOS VirtualBox 12/01/2006",
      "level": 4,
      "host": "vagrant-ubuntu-trusty-64",
      "facility": 0,
      "date": 1478654589000
    },
    "_score": 1,
    "_id": "syslog-tcp+0+75",
    "_type": "kafka-connect",
    "_index": "syslog-tcp"
  }
  ...
```

...おぉカーネルのログらしきものが入ってます！成功です。あっという間にパイプラインが出来上がりました。


# まとめ　〜Data Integreation as Service〜

Kafka-Connectの紹介と簡単な実験をして見ました。コネクタにより"関心の分離"が可能となり、システムが疎結合になりました。Data Integreation as Serviceへの重要な布石で「データはとりあえずKafkaに送っておけ」ば後はいかようにもなります。

次回はKafka-Streamsの紹介をしたいと思います。Kafka-StreamsはKafka-Connectと違い、双方向の通信が可能です。組み合わせると非常に強力かつ柔軟なパイプラインが構築できます。




# おまけ
**Confluent公認コネクタ**

| コネクタ | ソース・シンク | タグ | 開発元 | ダウンロード |
|-----------+-------------+------+-------------------+----------+
|HDFS | Sink | HDFS, Hadoop, Hive | Confluent |[Confluent](https://www.confluent.io/download/)|
|JDBC | Source | JDBC, MySQL | Confluent |[Confluent](http://docs.confluent.io/3.1.1/connect/connect-jdbc/docs/source_connector.html)|
|JDBC | Sink | JDBC, MySQL | Confluent |[Confluent](http://docs.confluent.io/3.1.1/connect/connect-jdbc/docs/sink_connector.html)|
|Attunity | Source | CDC | Attunity |[Attunity](http://www.attunity.com/products/replicate)|
|Couchbase | Source | Couchbase, NoSQL | Couchbase |[Couchbase](http://blog.couchbase.com/2016/september/kafka-connector-3-developer-preview-1)|
|DataStax | Sink | Cassandra, DataStax | Data Mountaineer |[Data Mountaineer](http://docs.datamountaineer.com/en/latest/cassandra-sink.html)|
|Elasticsearch | Sink | search, Elastic, log, analytics | Confluent |[Confluet](http://docs.confluent.io/3.1.1/connect/connect-elasticsearch/docs/elasticsearch_connector.html)|
|GoldenGate | Source | CDC, Oracle | Oracle |[Community](https://java.net/projects/oracledi/downloads/directory/GoldenGate/Oracle%20GoldenGate%20Adapter%20for%20Kafka%20Connect)|
|JustOne | Sink | Postgres | JustOne |[JustOne](https://github.com/justonedb/kafka-sink-pg-json)|
|Striim | Source | CDC, MS SQLServer, Oracle, MySQL | Striim |[Striim](http://www.striim.com/download-striim-for-confluent/)|
|Syncsort DMX | Source | DB2, IMS, VSAM, CICS | Syncsort |[Syncsort](http://www.syncsort.com/en/SupportandServices/SupportandServices)|
|Syncsort DMX | Sink | DB2, IMS, VSAM, CICS | Syncsort |[Syncsort](http://www.syncsort.com/en/SupportandServices/SupportandServices)|
|Vertica | Source | Vertica | HP Enterprise |[HP Enterprise](https://my.vertica.com/docs/7.2.x/PDF/HP_Vertica_7.2.x_Integrating_Apache_Kafka.pdf)|
|Vertica | Sink | Vertica | HP Enterprise |[HP Enterprise](https://my.vertica.com/docs/7.2.x/PDF/HP_Vertica_7.2.x_Integrating_Apache_Kafka.pdf)|
|VoltDB | Sink | VoltDB, NewSQL | VoltDB |[VoltDB](https://www.voltdb.com/download-confluentconnector)|

コミュティにより開発されたコネクタ

| コネクタ | ソース・シンク | タグ | 開発元 | ダウンロード |
|-----------+-------------+------+-------------------+----------+
|Apache Ignite | Source | File System | Community |[Community](https://github.com/apache/ignite/tree/master/modules/kafka)|
|Apache Ignite | Sink | File System | Community |[Community](https://github.com/apache/ignite/tree/master/modules/kafka)|
|Bloomberg Ticker | Source | Application feed | Community |[Community](http://docs.datamountaineer.com/en/latest/bloomberg.html)|
|Cassandra | Source | Cassandra | Community |[Community 1](http://docs.datamountaineer.com/en/latest/cassandra-source.html)|
|Cassandra | Sink | Cassandra | Community |[Community](https://github.com/tuplejump/kafka-connect-cassandra)|
|DynamoDB | ? | Dynamo, NoSQL | Community |[Community](https://github.com/shikhar/kafka-connect-dynamodb)|
|Elasticsearch | Sink | Elastic, search, log, analytics | Community |[Community 1](https://github.com/hannesstockner/kafka-connect-elasticsearch) [Community 2](https://github.com/ksenji/kafka-connect-es) [Community 3](https://github.com/DataReply/kafka-connect-elastic-search-sink)|
|FTP | Source | File System | Community |[Community](https://github.com/Eneco/kafka-connect-ftp)|
|Google PubSub | Source | Messaging | Community |[Community](https://github.com/GoogleCloudPlatform/pubsub/tree/master/kafka-connector)|
|Google PubSub | Sink | Messaging | Community |[Community](https://github.com/GoogleCloudPlatform/pubsub/tree/master/kafka-connector)|
|Hazelcast | Sink | Datastore, In-memory | Community |[Community](http://docs.datamountaineer.com/en/latest/hazelcast.html)|
|Hbase | Sink | Hbase, NoSQL | Community |[Community 1](https://github.com/mravi/kafka-connect-hbase) [Community 2](http://docs.datamountaineer.com/en/latest/hbase.html)|
|InfluxDB | Sink | Datastore, Time-series | Community |[Community](http://docs.datamountaineer.com/en/latest/influx.html)|
|Internet Relay Chat | Source | Application feed | Community |[Community](https://github.com/cjmatta/kafka-connect-irc)|
|Jenkins | Source | Application feed | Community |[Community](https://github.com/yaravind/kafka-connect-jenkins)|
|JMS | Sink | Messaging | Community |[Community](http://docs.datamountaineer.com/en/latest/jms.html)|
|Kudu | Sink | Kudu | Community |[Community](http://docs.datamountaineer.com/en/latest/kudu.html)|
|Mixpanel | Source | analytics | Community |[Community](https://github.com/cpard/kafka-connect-mixpanel)|
|MongoDB | Source | Mongo, MongoDB, NoSQL | Community |[Community](https://github.com/DataReply/kafka-connect-mongodb)|
|MongoDB CDC - Debezium | Source | MongoDB, CDC | Community |[Community](http://debezium.io/docs/connectors/mongodb/)|
|MQTT | Source | MQTT, messaging | Community |[Community](https://github.com/evokly/kafka-connect-mqtt)|
|MySQL CDC - Debezium | Source | MySQL, CDC, Oracle | Community |[Community](https://github.com/debezium)|
|Redis | Sink | Datastore, Messaging | Community |[Community](http://docs.datamountaineer.com/en/latest/redis.html)|
|RethinkDB | Source  | NoSQL, Streaming | Community |[Community](http://docs.datamountaineer.com/en/latest/rethink-source.html)|
|RethinkDB | Sink | NoSQL, Streaming | Community |[Community](http://docs.datamountaineer.com/en/latest/rethink.html)|
|Salesforce | Source | Datastore | Community |[Community](https://github.com/jcustenborder/kafka-connect-salesforce)|
|Solr | Sink | search, solr | Community |[Community 1](https://github.com/MSurendra/kafka-connect-solr/) [Community 2](https://github.com/jcustenborder/kafka-connect-solr)|
|Solr | Source | search, solr | Community |[Community](https://github.com/jcustenborder/kafka-connect-solr)|
|Splunk | Sink | Datastore, log | Community |[Community](https://github.com/jcustenborder/kafka-connect-splunk)|
|Splunk | Source | Datastore, log | Community |[Community](https://github.com/jcustenborder/kafka-connect-splunk)|
|spooldir | Source | File System | Community |[Community](https://github.com/jcustenborder/kafka-connect-spooldir)|
|SQS | Source | AWS, SQS, Messaging | Community | [Community](https://github.com/ConnectedHomes/sqs-kafka-connect)|
|Syslog | Source | log | Community |[Community](https://github.com/jcustenborder/kafka-connect-syslog)|
|S3 | Sink | AWS, S3, log | Community | [Community 1](https://github.com/DeviantArt/kafka-connect-s3) [Community 2](https://github.com/qubole/streamx)|
|Twitter | Source | Application feed | Community |[Community 1](https://github.com/jcustenborder/kafka-connect-twitter) [Community 2](https://github.com/Eneco/kafka-connect-twitter)|
|Twitter | Sink | Application feed | Community | [Community](https://github.com/Eneco/kafka-connect-twitter)|