TwitterとAiven for Apache Kafka®で独自のコネクタを使用する

<font color="Grey">*[Use your own connector with Twitter and Aiven for Apache Kafka®](https://aiven.io/blog/connecting-twitter-to-aiven-for-apache-kafka)の翻訳です。*</font>

2022年5月24日

# Apache Kafka®のためのTwitterとAivenで独自のコネクタを使用する

Aivenがあなたの望むApache Kafka®コネクターを提供していない場合はどうしますか？TwitterのメッセージをKafkaに収集するために、外部コネクターを使用する方法を学びましょう。

## Aiven for Apache Kafka®で独自のコネクタを使用する - Twitterソースの例


Apache Kafka®は、より多くの企業にとって中心的なデータバックボーンとなっており、豊富なApache Kafka®コネクトエコシステムがさまざまなプラグインを提供することで、Kafkaと多種多様なテクノロジーを簡単に統合することができます。


Apache Kafkaをセルフホスティングする場合、適切なプラグインと設定オプションを見つけることで、どのようなオープンソースのコネクタープラグインでも利用することができます。マネージドサービスを利用する場合、状況は変わります。サポートされているプラグインのリストが驚くほど多くても（[Aivenが提供するすべてのプラグイン](https://developer.aiven.io/docs/products/kafka/kafka-connect/concepts/list-of-connector-plugins.html)をご覧ください）、あなたが探していた特定の`sink-to-obscure-datatech`が見つからないかもしれません。それでも、Aivenは提案を受け入れ、リストに含めるために継続的に新しいコネクタプラグインを評価します！


また、自己管理型のApache Kafka® Connectクラスタを作成し、Aiven for Apache Kafkaに接続することもできます。この方法では、Aivenが提供するマネージドKafkaサービスの恩恵を受けながら、オープンソースのコネクターを自由に選択することができます。[弊社開発者ポータル](https://developer.aiven.io/docs/products/kafka/kafka-connect/howto/bring-your-own-kafka-connect-cluster.html)でプロセスの例を読むことができます。このブログポストでは、この町で最高のコネクタの1つ、おそらく新規学習者の90%が使用しているコネクタ、[Twitterソースコネクタ](https://github.com/jcustenborder/kafka-connect-twitter)の使用を開始するために必要な手順を説明します。


### Apache Kafkaクラスタの作成


[Aiven CLI](https://developer.aiven.io/docs/tools/cli.html)と専用の `service create` 関数を使って、この部分を手短に説明しよう。利用可能なすべてのパラメータは[専用ページ](https://developer.aiven.io/docs/tools/cli/service.html#avn-service-create)をご覧ください。また、すべての[高度なカスタマイズパラメータ](https://developer.aiven.io/docs/products/kafka/reference/advanced-params.html)のリストもご覧ください。Aivenが提供しています。今回のブログでは以下のパラメータを使用します：



```
AVN SERVICE CREATE DEMO-KAFKA
 --service-type kafka
 --cloud google-europe-west3  \
 --plan business-4 ╱ -c kafka.auto_create_topics_enable
 -c kafka.auto_create_topics_enable=true ￤ -c kafka_rest=true
 -c kafka_rest=true` クリップボードにコピーする
```

上記のコマンドは、`google-europe-west3`リージョンに`business-4`プランのAiven for Apache Kafkaインスタンス`demo-kafka`を作成します。また、トピックとREST APIの自動作成も有効にしている。このAPIは、トピックに格納されたデータを確認するために使用する。


サービスの開始中に、ローカルの Apache Kafka Connect クラスタを `demo-kafka` サービスに統合するために使用する [Java keystore and truststore](https://developer.aiven.io/docs/products/kafka/howto/keystore-truststore) を生成することができる。  

以下のAiven CLIコマンドで両方のストアを作成できる：



```
avn service user-kafka-java-creds demo-kafka ˶ -d certsfolder
 -d certsfolder
 -p STOREPASSWORD123
 --ユーザー名 avnadmin`Copy to clipboard
```

上記は必要な証明書を `certsfolder` という名前のフォルダにダウンロードし、同じフォルダに `client.keystore.p12` という名前のキーストアファイルと `client.truststore.jks` という名前のトラストストアを作成します。ストアとキーのセキュリティのために異なるシークレットを設定することに熱心であれば、[開発者専用ポータル ドキュメント](https://developer.aiven.io/docs/products/kafka/howto/keystore-truststore) を確認するとよいだろう。


### 自己管理型の Apache Kafka Connect クラスタを作成する。


さて、シェルのスキルを使う時が来た。唯一の前提条件は、JDKがインストールされていることだ。まずはApache Kafkaのバイナリを入手することから始めよう。



```
wget https://dlcdn.apache.org/kafka/3.1.0/kafka_2.13-3.1.0.tgz`Copy to clipboard
```

次に解凍する。



```
tar -xzf kafka_2.13-3.1.0.tgz` クリップボードにコピーする。
```

これで、すべての Apache Kafka グッズが入った `kafka_2.13-3.1.0` というフォルダが作成される。


#### Twitterソースコネクターの依存関係を追加する


Twitterデータの取得を開始するには、[専用オープンソースコネクター](https://github.com/jcustenborder/kafka-connect-twitter)を使用します。関連するコードは



```
wget https://github.com/jcustenborder/kafka-connect-twitter/releases/download/0.2.26/kafka-connect-twitter-0.2.26.tar.gz`Copy to clipboard
```

tarファイルを解凍する



```
mkdir twitter-connector
tar -xvf kafka-connect-twitter-0.2.26.tar.gz -C twitter-connector`クリップボードにコピーする
```

上記のコマンドを実行すると、tarファイルが解凍され、`twitter-connector`というフォルダに格納される。このフォルダには、Apache Kafka Connectクラスタが読み込む必要があるすべてのファイルが格納された`usr/share/kafka-connect/kafka-connect-twitter`というサブフォルダがある。これらのファイルは `kafka_2.13-3.1.0` フォルダ内の `plugin` サブフォルダに移動できる。



```
`mkdir kafka\_2.13-3.1.0/plugins
mv twitter-connector/usr/share/kafka-connect/kafka-connect-twitter kafka_2.13-3.1.0/plugins/lib``Copy to clipboard
```

#### Apache Kafka Connect設定ファイルを定義する


次に、ローカルのKafka ConnectクラスタがAiven for Apache Kafkaを指すようにするための設定ファイルを定義します。[開発者ポータルテンプレート](https://github.com/aiven/devportal/blob/main/code/products/kafka/my-connect-distributed.properties)を使用して、`my-connect-distributed.properties`という名前のファイルを作成し、代入します：


* PATH_TO_KAFKA_HOME` を Apache Kafka のバイナリがあるディレクトリに置き換える。
* APACHE_KAFKA_HOST:APACHE_KAFKA_PORT` を `demo-kafka` のホスト名とポートに置き換える。



```
avn service get demo-kafka --format '{service_uri}'`Copy to clipboard
```

* TRUSTSTORE_PATH` と `KEYSTORE_PATH` には、キーストアとトラストストアのファイルが格納されているフォルダのパスを指定する（前のセクションで定義したコマンドを使用した場合は `certsfolder`）。
* KEY_TRUST_SECRET` にキーストアとトラストストアの秘密鍵 (前のセクションで定義したコマンドを使用した場合は `STOREPASSWORD123`) を指定する。


#### ローカルのConnectクラスタを起動する。


設定と必要なファイルがすべて揃ったので、ローカルのApache Kafka Connectクラスタを次のように起動します：



```
./kafka_2.13-3.1.0/bin/connect-distributed.sh ./my-connect-distributed.properties`Copy to clipboard
```

### Twitterへのアクセスを設定する


Apache Kafka Connectクラスタが稼働しているので、[Twitterの開発者ポータルページ](https://developer.twitter.com/)の**開発者ポータル**にアクセスして、コネクターがツイートのソーシングを開始するために必要な認証情報を提供する新しいアプリケーションを作成します。


1.メインダッシュボードのページで、v2エンドポイントを使用して新しいプロジェクトを作成します。プロジェクトには、以下を指定する必要があります：


	* プロジェクト名
	* APIの探索**、ボットの作成**、コンシューマ・ツールの作成**など、さまざまな選択肢の中から、目的に合ったものを選ぶことができます。
	* プロジェクトの説明では、目的の概要を説明することができます。
	* 既存のアプリを使うのか、新しく作るのかを定義します。私たちはTwitter APIの初心者なので、新しいアプリを作成します。
2.App Setup*セクションでは、新しいアプリの設定を定義します：


	* アプリの**環境**、ブログ投稿の目的では*Development*を選択します。
	* アプリに覚えやすい名前を付けることで、そのアプリが何のために使われているのかを追跡することができます。twitter-kafka-connect-<SUFFIX>`と呼ぶことができます。ここで、`<SUFFIX>`は一意な識別子である必要があります（すべてのアプリ名は一意である必要があります）。
	* アプリの**キーとトークン**は、必要なキーを生成して取り出すことができる。このセクションから、後にApache Kafkaコネクタのセットアップで使用する**API Key**と**API Key Secret**をコピーする必要がある（これらを`TWITTER_API_KEY`と`TWITTER_API_SECRET`として参照する）。すべてのセットアップが完了したら、**App Settings**を選択すると、アプリとプロジェクトが正常にビルドされたことを伝える以下のような画面が表示されるはずだ。


![Twitter App and Project build](https://cdn.sanity.io/images/sczeoy4w/production/059e6c82eb846d62ddf16c8a284a8b684785e7c9-1314x1025.jpg?q=80&fit=min&auto=format&dpr=1.5)
3.次に、**Keys and tokens** タブに移動して、コネクタの動作に必要な追加シークレットを生成（または再生成）します。
4.アクセス・トークンとシークレットを生成します。


	* アクセストークン: `TWITTER_ACCESS_TOKEN`
	* アクセストークン・シークレット： `TWITTER_ACCESS_TOKEN_SECRET`
5.Twitter開発者ポータルでの最後の設定は、私たちのプロジェクトのために**Elevated**アクセスをリクエストすることです。Twitter API v2がリリースされたため、デフォルトの(**Essential**)アクセスではv2としかやり取りできません。コーディング・スキル・レベルやプロジェクトの説明を含むいくつかの情報を入力し、Developer Agreementに同意する必要がある。リクエストはTwitterによって公式にチェックされ、すべてがうまくいけば、数時間かかることもあるが、すぐに昇格アカウントを手に入れることができるだろう。


#### ソースコネクタ設定ファイルの作成


上記で取得したツイッターの秘密を、`twitter-source.json`という名前の設定ファイルに以下の内容で記述する。



```
`{
 "name": "twitter_connector"、
 "config"：
 {
 "tasks.max": "1"、
 "connector.class": "com.github.jcustenborder.kafka.connect.twitter.TwitterSourceConnector"、
 "process.delees": "false"、
 "filter.keywords": "データベース"、
 "kafka.status.topic": "twitter-topic"、
 "twitter.auth.consumerKey": "TWITTER_API_KEY"、
 "twitter.oauth.consumerSecret": "TWITTER_API_SECRET"、
 "twitter.oauth.accessToken": "TWITTER_ACCESS_TOKEN"、
 "twitter.oauth.accessTokenSecret": "TWITTER_ACCESS_TOKEN_SECRET"
 }
クリップボードにコピーする
```

以下はカスタマイズ可能なパラメータである：


* `"process.deletes": "false"`: 簡単のため、ツイートの削除を処理しない。
* `"filter.keywords": "database"`: `database`キーワードを含むツイートをフィルタリングする。このキーワードは、Apache Kafkaにデータが表示されるように、ツイートの中で発生する、読んだ瞬間に話題となるものに置き換えることをお勧めします。
* `"kafka.status.topic": "twitter-topic"`: 対象のトピックは `twitter-topic` という名前になる。
* `twitter.oauth` パラメータの値は、前のステップで取得した twitter の秘密情報で変更する必要がある。


#### Twitter ソースコネクタを起動する


[Apache Kafka REST API](https://kafka.apache.org/documentation.html#connect_rest)を使ってコネクタを起動します：



```
``curl -s -H "Content-Type: application/json" -X POST ˶ -d @twitter-source.json ˶.
 -d @twitter-source.json \
 http://localhost:8083/connectors/`クリップボードにコピー
```

#### Apache Kafkaで出力を確認する


上記の `curl` コマンドは、`twitter-source.json` コネクタ設定ファイルを渡す `connectors` REST エンドポイントを使用する。これで、`twitter-topic` Apache Kafka トピックに `apachekafka` を含むツイートが流れてくるのが確認できるはずだ。KarapaceのREST APIを有効にしたので、[Aiven Console](https://console.aiven.io/)にアクセスし、*Topic*タブの下にある`demo-kafka`サービス名をクリックすることで、トピックデータを確認することができる。


![Aivenコンソールに表示されるデータベースに関するツイート](https://cdn.sanity.io/images/sczeoy4w/production/74da099db6ce98edfefc2cf20c7032954db19db5-1330x1233.jpg?q=80&fit=min&auto=format&dpr=1.5)


これはストリーミング・ソリューションなので、定義したキーワードを含むツイートが書き込まれた場合のみ、トピックにデータが表示されることに注意してください。コネクターをテストしたい場合は、キーワードを含むツイートを自分で書いてみてください！


ビッグニュース：現在、Aiven for Apache KafkaでAivenがサポートしていないApache Kafka® Connectorプラグインを使用しています！Apache Kafka Connectクラスタを管理する負担はありますが、私たちのソース/ターゲット技術が[利用可能なコネクタ](https://developer.aiven.io/docs/products/kafka/kafka-connect/concepts/list-of-connector-plugins.html)のどれにも対応していない場合に最適なオプションかもしれません。


クイック・ヒント: コネクタの設定を変更する必要がある場合は、まずコネクタを



```
curl -s -X DELETE http://localhost:8083/connectors/twitter_connector`Copy to clipboard
```

そして、更新された設定ファイルを `curl` コマンドで `PUSH` オプションを指定して送信します。


### マネージド Apache Kafka サービスとコネクタプラグイン選択の自由度


Apache Kafkaクラスタを完全に管理するのは面倒な作業であり、AivenのようなマネージドサービスをApache Kafkaのために使用することは、通常、賢明なアイデアです。特殊なユースケースの場合、探している構成やコネクタがAivenでサポートされていないことに気づくかもしれません。Aiven for Apache Kafkaを使用すれば、特定の統合問題を解決するローカルのApache Kafka Connectクラスタを簡単に統合でき、全体的なApache Kafkaマネージドソリューションと豊富なオープンソースコネクタの両方のメリットを享受できます。


詳しくは以下のリソースをご覧ください：


* [サポートされている Apache Kafka Connect プラグインのリスト](https://developer.aiven.io/docs/products/kafka/kafka-connect/concepts/list-of-connector-plugins.html)
* [Aiven for Apache Kafka 詳細設定パラメータ](https://developer.aiven.io/docs/products/kafka/reference/advanced-params.html) のリスト。
* [Twitterソースコネクタドキュメント](https://github.com/jcustenborder/kafka-connect-twitter)
* [ローカルJDBCシンクコネクタのセットアップ](https://developer.aiven.io/docs/products/kafka/kafka-connect/howto/bring-your-own-kafka-connect-cluster.html)