Power Grid Data

Ansible内で使用する変数を平文でファイルに記載し、プレイブック集に入れ込むのに不安を感じるときがある。そのようなとき、Ansible Vaultを利用すると暗号化された状態で変数を管理できる。 Ansible Vaultで管理された変数をプレイブック内で通常の変数と同様に扱えるため見通しが良くなる。

公式サイトの Encrypting content with Ansible Vault が網羅的でわかりやすいのだが、具体例が足りない気がしたので以下に一例を示しておく。ここに挙げた以外の使い方は、公式サイトを参照されたし。

Ansible-Vaultを用いた機密情報の暗号化のTips も参考になった。

暗号化されたファイルの作成

secret.yml 内に変数を記述し、暗号化する。後ほど暗号化されたファイルを変数定義ファイルとして読み込む。

ここでは以下のような内容とする。

secret.yml

1 2	hoge: fuga: foo

ファイルを暗号化する。

1	$ ansible-vault create secret.yml

上記ファイルを作成する際、Vault用パスワードを聞かれるので適切なパスワードを入力すること。あとでパスワードは利用する。

（参考）復号して平文化

1	$ ansible-vault decrypt secret.yml

（参考）暗号化されたファイルの編集

1	$ ansible-vault edit secret.yml

（参考）Vaultパスワードをファイルとして渡す

プロンプトで入力する代わりに、どこかプレイブック集の外などにVaultパスワードを保存し利用することもできる。ここでは、 ~/.vault_password にパスワードを記載したファイルを準備したものとする。編集する例を示す。

1	$ ansible-vault edit secret.yml --vault-password-file ~/.vault_password

プレイブック内で変数として利用

変数定義ファイルとして渡し、プレイブック内で変数として利用する。以下のようなプレイブックを作る。

test.yml

- hosts: localhost
  vars_files:
    - secret.yml
  tasks:
    - name: debug
      debug:
        msg: "{{ hoge.fuga }}"

以下、簡単な説明。

変数定義ファイルとして secret.yml を指定（ vars_files の箇所）
今回はdebugモジュールを利用し、変数内の値を表示することとする。
暗号化された変数定義ファイル secret.yml に記載されたとおり、構造化された変数 hoge.fuga の値を利用する。
結果として、foo という内容がSTDOUTに表示されるはず。

プレイブックを実行する。

1	$ ansible-playbook test.yml --ask-vault-pass

Use static image on Hexo blog

2020-12-31

Hexo

参考
メモ

参考

メモ

Hexo 記事に画像を貼り付けるを参考に、もともとCacooのリンクを使っていた箇所をすべてスタティックな画像を利用するようにした。

post_asset_folderを利用して、記事ごとの画像ディレクトリを利用することも考えたが、画像はひとところに集まっていてほしいので、 Global-Asset-Folder を利用することにした。

なお、上記ブログでは

1	プロジェクトトップ/images/site

以下に画像を置き、

1	![猫](/images/site/cat.png)

のようにリンクを指定していた。

自身の環境では、rootを指定しているので

1	プロジェクトトップ/image

以下にディレクトリを置き、

1	![猫](memo-blog/images/cat.png)

と指定することにした。

Delta Lake with Alluxio

2020-12-31

Storage LayerDelta Lake

Alluxio, Delta Lake

参考
メモ
- 疑似分散で動作確認

参考

論文

Hadoop環境

ansible-bigdata

Alluxioドキュメント

メモ

Delta Lake High-Performance ACID Table Storage over Cloud Object Stores や Lakehouse A New Generation of Open Platforms that Unify Data Warehousing and Advanced Analytics の論文の通り、 Delta Lakeはキャッシュとの組み合わせが可能である。

今回は、ストレージにHDFS、キャッシュにAlluxioを使って動作確認する。

疑似分散で動作確認

実行環境の準備

ansible-bigdata あたりを参考に、Hadoopの疑似分散環境を構築する。 Bigtopベースの2.8.5とした。

併せて、同Ansibleプレイブック集などを用いて、Spark3.1.1のコミュニティ版を配備した。

併せて、Alluxioは2.5.0-2を利用。

Alluxioに関しては、以下のようにコンパイルしてパッケージ化して用いることできる。

$ sudo -u alluxio mkdir /usr/local/src/alluxio
$ sudo -u alluxio chown alluxio:alluxio /usr/local/src/alluxio
$ cd /usr/local/src/alluxio
$ sudo -u alluxio git clone git://github.com/alluxio/alluxio.git
$ sudu -u alluxio git checkout -b v2.5.0-2 refs/tags/v2.5.0-2
$ sudo -u alluxio mvn install -Phadoop-2 -Dhadoop.version=2.8.5 -DskipTests

コンフィグとしては以下を利用。

$ cat conf/alluxio-site.properties

(snip)

# Common properties
# alluxio.master.hostname=localhost
alluxio.master.hostname=localhost
# alluxio.master.mount.table.root.ufs=${alluxio.work.dir}/underFSStorage
# alluxio.master.mount.table.root.ufs=/tmp
alluxio.master.mount.table.root.ufs=hdfs://localhost:8020/alluxio

# Security properties
# alluxio.security.authorization.permission.enabled=true
# alluxio.security.authentication.type=SIMPLE
alluxio.master.security.impersonation.yarn.users=*
alluxio.master.security.impersonation.yarn.groups=*

# Worker properties
# alluxio.worker.ramdisk.size=1GB
alluxio.worker.ramdisk.size=1GB
# alluxio.worker.tieredstore.levels=1
# alluxio.worker.tieredstore.level0.alias=MEM
# alluxio.worker.tieredstore.level0.dirs.path=/mnt/ramdisk

# User properties
# alluxio.user.file.readtype.default=CACHE
# alluxio.user.file.writetype.default=MUST_CACHE

ポイントは以下の通り。

疑似分散環境のHDFS利用
Alluxioの使用するディレクトリとして、/alluxioを利用
マスタはローカルホストで起動

フォーマット、起動

1 2	$ sudo -u alluxio ./bin/alluxio format $ sudo -u alluxio ./bin/alluxio-start.sh local SudoMount

テストを実行

1	$ sudo -u alluxio ./bin/alluxio runTests

もしエラーが生じた場合は、例えばHDFSの/alluxioディレクトリに、適切な権限設定、所有者設定がされているかどうかを確認すること。

Alluxioが起動すると以下のようなUIを確認できる（ポート19999）

先程テストで書き込まれたファイル群が見られるはず。

ここでは、上記の通り、環境を整えた前提で以下説明する。

Sparkの起動確認

Examples: Use Alluxio as Input and Output を参考に、Alluxio経由での読み書きを試す。

予め、今回の動作確認で使用するテキストデータ（AlluxioのREADME）をアップロードしておく。

1	$ sudo -u alluxio /opt/alluxio/default//bin/alluxio fs copyFromLocal /opt/alluxio/default/LICENSE /LICENSE

予め、以下のような設定をspark-defaults.confに入れておく。 Alluxioのクライアントライブラリを用いられるように。

1 2	spark.driver.extraClassPath /opt/alluxio/default/client/alluxio-2.5.0-2-client.jar spark.executor.extraClassPath /opt/alluxio/default/client/alluxio-2.5.0-2-client.jar

Sparkが起動することを確認する。ここではDelta Lakeも含めて起動する。

$ /usr/local/spark/default/bin/spark-shell \
  --packages io.delta:delta-core_2.12:0.8.0 \
  --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
  --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"

起動したシェルでAlluxio上のREADMEファイルを読み取り、行数を確認する。

scala> val pathOnAlluxio = "alluxio://localhost:19998/LICENSE"

scala> val testDF = spark.read.text(pathOnAlluxio)

scala> testDF.count
res0: Long = 482

Delta Lakeを通じて書き込む動作確認

準備として、Alluxio上に、dobachiユーザ用のディレクトリを作成してみる。

1
2
3

$ sudo -u alluxio /opt/alluxio/default/bin/alluxio fs mkdir /users
$ sudo -u alluxio /opt/alluxio/default/bin/alluxio fs mkdir /users/dobachi
$ sudo -u alluxio /opt/alluxio/default/bin/alluxio fs chown dobachi:dobachi /users/dobachi

先程起動しておいたシェルで、Delta Lake形式のデータを書き込んで見る。

scala> val data = spark.range(0, 5)
data: org.apache.spark.sql.Dataset[Long] = [id: bigint]

scala> val outputUrl = "alluxio://localhost:19998/users/dobachi/numbers"

scala> data.write.format("delta").save(outputUrl)

すると以下のようなエラーが生じた。

scala> data.write.format("delta").save(outputUrl)
21/01/05 22:47:50 ERROR HDFSLogStore: The error typically occurs when the default LogStore implementation, that
 is, HDFSLogStore, is used to write into a Delta table on a non-HDFS storage system.
 In order to get the transactional ACID guarantees on table updates, you have to use the
 correct implementation of LogStore that is appropriate for your storage system.
 See https://docs.delta.io/latest/delta-storage.html " for details.

org.apache.hadoop.fs.UnsupportedFileSystemException: fs.AbstractFileSystem.alluxio.impl=null: No AbstractFileSystem configured for scheme: alluxio

(snip)

当たり前だが、Delta Lakeの下回りのストレージとして標準では、 Alluxioが対応しておらず、LogStoreからエラーが生じた、ということのようだ。

一瞬、LogStoreを新たに開発しないといけないか？と思ったものの、よく考えたら、HDFSHadoopFileSystemLogStoreから Alluxioのスキーマを認識させてアクセスできるようにすればよいだけでは？と思った。そこで、Hadoopの設定でAlluxioFileSystemをalluxioスキーマ（ファイルシステムのスキーマ）に明示的に登録してみる。

/etc/hadoop/conf/core-site.xmlに以下を追記。

<property>
  <name>fs.AbstractFileSystem.alluxio.impl</name>
  <value>alluxio.hadoop.AlluxioFileSystem</value>
  <description>The FileSystem for alluxio uris.</description>
</property>

再びSparkを立ち上げ、適当なデータを書き込み。

scala> val data = spark.range(0, 5)

scala> val outputUrl = "alluxio://localhost:19998/users/dobachi/numbers"

scala> data.write.format("delta").save(outputUrl)

以上のように書き込みに成功した。

つづいて、テーブルとして読み出す。

scala> val df = spark.read.format("delta").load(outputUrl)
scala> df.show
+---+
| id|
+---+
|  1|
|  2|
|  3|
|  4|
|  0|
+---+

テーブルへの追記。

scala> val addData = spark.range(5, 10)
scala> addData.write.format("delta").mode("append").save(outputUrl)
scala> df.show
+---+
| id|
+---+
|  3|
|  9|
|  6|
|  5|
|  1|
|  8|
|  4|
|  2|
|  7|
|  0|
+---+

また、追記書き込みをしたのでDeltaログが増えていることが分かる。（3回ぶんのログがあるのは、↑には記載していないがミスったため）

（補足）HDFS上のディレクトリ権限に関するエラー

Sparkでの処理実行時にYARNで実行していたところ、Executorにおける処理からAlluxioを呼び出すときにエラー。 yarnユーザでのアクセスとなり、HDFS上の /alluxio へのアクセス権がなかったと考えられる。

1
2

21/01/05 02:54:35 WARN TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0, hadoop-pseudo, executor 2): alluxio.exception.status.UnauthenticatedException: Channel authentication failed with code:UNAUTHENTICATED. Channel: GrpcChannelKey{ClientType=FileSystemMasterClient, ClientHostname=hadoop-pseudo.mshome.net, ServerAddress=GrpcServerAddress{HostName=localhost, SocketAddress=localhost:19998}, ChannelId=81f7d97f-8e32-4289-bcab-ea6008d5ffac}, AuthType: SIMPLE, Error: alluxio.exception.status.UnauthenticatedException: Plain authentication failed: Failed to authenticate client user="yarn" connecting to Alluxio server and impersonating as impersonationUser="vagrant" to access Alluxio file system. User "yarn" is not configured to allow any impersonation. Please read the guide to configure impersonation at https://docs.alluxio.io/os/user/2.4/en/operation/Security.html
        at alluxio.exception.status.AlluxioStatusException.from(AlluxioStatusException.java:141)

Alluxio Security のドキュメント中に「Client-Side Hadoop Impersonation」を読むと、「なりすまし」を許可する設定があるようだ。

そこで、yarnユーザが様々なユーザになりすませるような簡易設定を以下のように加えることにした。実際の運用する際は、なりすましのスコープに注意したほうが良さそうだ。

conf/alluxio-site.properties

1 2	alluxio.master.security.impersonation.yarn.users=* alluxio.master.security.impersonation.yarn.groups=*

ドキュメントではクライアントで alluxio.security.login.impersonation.username も指定するよう書かれていたが、起動時にしてしなくてもアクセスできるようになった。あとで実装を調べたほうが良さそうだ。

Information about trends in 2020

2020-12-26

ResearchTrends

Trends

参考
メモ

参考

ガートナー

ガートナー、「先進テクノロジのハイプ・サイクル：2020年」を発表

データ・ファブリック

DNAコンピューティング

アダプティブラーニング

米軍の教育でもAIが活躍　「アダプティブ・ラーニング」の可能性 (1/4)

デジタル・ツイン

IOWN

メモ

オープンになっている情報を中心に既存情報を軽くまとめる。特にデータ処理、データ活用との関連を探る。

先進テクノロジのハイプ・サイクル：2020年

ガートナー、「先進テクノロジのハイプ・サイクル：2020年」を発表の記事によると、 2020/8/19にハイプ・サイクルが発表された。

記事冒頭で触れられているとおり、COVID-19の影響は否定できない。例ではヘルス・パスポート、ソーシャル・ディスタンシング・テクノロジが挙げられていた。記載されているととおり、「過度な期待」として初登場する技術は少ないのだが、それに該当するということで注目されている。

挙げられていたトレンドと所感

デジタル・ミー
- つまりは、デジタル・ツイン
- 多分にもれずデジタル・ツイン関連は取り上げられていた。
- 過去になかったデータを使うようになるという側面はあるが、それに合わせた新たなアーキテクチャが必要になるか。
  - 過去のブーム面との例
    - 大量の顧客行動情報、ログ：非構造データ処理に適した並列分散処理フレームワーク（MapReduceなど）
    - 大量の画像データ：深層学習
    - 大量のセンサーデータ：IoT、ストリーム処理
    - 大量のデジタル・ツイン・データ：？
コンポジット・アーキテクチャ
- 一言で言うとデータ・ファブリック※1を基礎としたアーキテクチャ
  - ※1）つまりオンプレ、クラウドに点在するデータを最適な場所に配置し、いつでも利用可能にする
- 重要なのは柔軟性を実現するという目論見
- 柔軟性を求める組織に必要とされるアーキテクチャと考えることもできそうだ
フォーマティブAI
- 一言で言うと状況に動的に対応するAI
- 動的に適応していく、さらに突き詰めると「自律的に…」というニュアンスが生じそう
- AIに関する倫理の議論と関連すると考えられる。
アルゴリズムによる信頼
- 一言で言うと、責任ある権限に基づく信頼モデルから、アルゴリズムによる信頼への転換
シリコンの先へ
- 一言で言うと、シリコンに変わる新素材に対する期待
- 参考
  - DNAコンピューティング（日本生物理学会）、DNAコンピューティングの原理と展開（小倉裕介）

個人的にピックアップしたキーワードと所感

アダプティブな機械学習
- 「学習」という領域におけるパーソナライズの一つの形と思えばよいか
- 近年ではAIの普及もあり、機械的な支援をもとに実現することも増えてきたか。
- 参考
  - 米軍の教育でもAIが活躍　「アダプティブ・ラーニング」の可能性 (1/4)
シチズン・ツイン
- Google検索では適切な結果は得られなかった
人のデジタル・ツイン
- シチズン・ツインと同様、モノのデジタル化に続いて、ヒトのデジタル化を目指す、ということか。
- NTTサービスイノベーション総合研究所の提唱する「デジタルツインコンピューティング」では、層の中間層を設けることが重要とされている。
- 参考
- ヒトと社会のデジタル化世界 ─ デジタルツインコンピューティング ─
- IOWNでヒトもデジタル化　“私のコピー”が仮想空間で仕事する世界
データ・ファブリック
- 「オンプレ、クラウドに点在するデータを最適な場所に配置し、いつでも利用可能にする」というコンセプトは特徴的だが、ベンダによって実際の定義が異なるように感じる。
- 「データ民主化」と同様のコンセプトに感じる。
- ポイントはストレージとしての特徴ではなく、「統合」を実現する側面か
- 参考情報
  - データファブリックを実現するクラウドデータサービス、ネットアップとNEC、国内初となるデータファブリックソリューションの提供を開始、 NetAppのデータファブリックは時代遅れなのか？、世界のデータファブリックの市場、Talend Data Fabric、 Splunk エンタープライズマシンデータファブリック

週刊ダイヤモンド 2020年12/12号

週刊ダイヤモンド 2020年12/12号の中に、「NTT帝国の逆襲」という特集記事がある。

技術というより、事業職が強いが念の為に（あえて）技術に関連するキーワードだけ述べる。なお、この書籍で重要なのは、本来はどちらかというとビジネス戦略側面である。詳細は当該書籍を参照されたし。

技術キーワード

IOWN構想
- 光ファイバー、ネットワーク転送、音声等のNTT特許数は圧倒的である。
- 電力効率100話い、伝送容量125倍、遅延を200分の1を目指す
- 光電子融合型の光トランジスタ開発によりリード。
O-RAN
- 通信規格を主導。オープン化で囲い込みを打破する。
V-RAN
- 通信の仮想化

IOWN公開情報

あくまで公開された情報ではあるがリスト化しておく。

IOWN公式ウェブサイト
- 概要や目指しているところを理解できるコンテンツが公開されている
- 構成要素は以下の通り。
  - 「オールフォトニクス・ネットワーク」
    - 「大容量光伝送システム・デバイス」、「イジングマシン」、「光格子時計ネットワーク」
  - 「デジタルツインコンピューティング」
    - 「多様な産業やモノとヒトのデジタルツインを自在に掛け合わせて演算を行う」ことが特色
  - 「コグニティブ・ファウンデーション」
    - ICTリソースを全体最適に調和させ、必要な情報をNW内に流通させる。

Reference of connector plugin

2020-12-16

Messaging SystemKafka

Kafak Connect, Kafka

参考
メモ
- kafka-connect-syslog
  - 動作確認
- kafka-connect-datagen

参考

メモ

2020/12時点で、Kafka Connectのプラグインの参考になるもの探す。

kafka-connect-syslog

kafka-connect-syslog が最も簡易に動作確認できそうだった。ただ、 ConfluentのGitHub を見る限り、GitHub上には実装が公開されていないようだった。

動作確認

ここでは、ローカルに簡易実験用の1プロセスのKafkaを起動した前提とする。起動方法は Kafkaのクイックスタートを参照。

kafka-connect-syslog のパッケージをダウンロードして /opt/connectors 以下に展開。

1
2
3

$ cd /opt/connectors
$ sudo unzip confluentinc-kafka-connect-syslog-1.3.2.zip
$ sudo chown -R kafka:kafka confluentinc-kafka-connect-syslog-1.3.2

というパッケージが展開される。

動作確認に使用するプロパティは以下。

$ cat etc/minimal-tcp.properties
#
# Copyright [2016 - 2019] Confluent Inc.
#

name=syslog-tcp
tasks.max=1
connector.class=io.confluent.connect.syslog.SyslogSourceConnector
syslog.port=5515
syslog.listener=TCP
confluent.license
confluent.topic.bootstrap.servers=localhost:9092
confluent.topic.replication.factor=1

また、Connectの設定には以下を追加する。

/opt/kafka_pseudo/default/config/connect-standalone.properties

1
2
3

(snip)

plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,

プラグインを置く場所として、 /opt/connectors を指定した。

/opt/kafka_pseudo/default/bin/connect-standalone.sh を利用して、スタンドアローンモードでKafka Connectを起動。

1
2

$ sudo -u kafka /opt/kafka_pseudo/default/bin/connect-standalone.sh /opt/kafka_pseudo/default/config/connect-standalone.properties \
  /opt/connectors/confluentinc-kafka-connect-syslog-1.3.2/etc/minimal-tcp.properties

起動したのを確認し、別の端末から適当なデータを送信。

1	$ echo "<34>1 2003-10-11T22:14:15.003Z mymachine.example.com su - ID47 - Your refrigerator is running" \| nc -v -w 1 localhost 5515

Console Consumerを利用して書き込み状況を確認。

1 2	$ cd ${KAFKA_HOME} $ ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic syslog --from-beginning

先程書き込んだものが表示されるはずである。

kafka-connect-datagen

kafka-connect-datagen もあった。 kafka-connect-datagenのGitHub に実装も公開されているように見える。ドキュメントのリンクから、当該レポジトリのREADMEにジャンプしたため、そのように判断。

以降、v0.4.0を対象として確認したものである。

概要

指定されたスキーマで、ダミーデータを生成するコネクタ。 avro-random-generator を内部的に利用している。

スキーマ指定はAvroのスキーマファイルを渡す方法もあるし、組み込みのスキーマを指定する方法もある。 kafka-connect-datagenのサンプルスキーマを参照。

また、Kafkaに出力する際のフォーマットは指定可能。 Kafka Connect自体の一般的なパラメータである、 value.converter を指定すれば良い。例えば以下の通り。

1	"value.converter": "org.apache.kafka.connect.json.JsonConverter",

実装状況

2020/12/21時点では本プロジェクトのバージョンは0.4.0であり、

<parent>
    <groupId>io.confluent</groupId>
    <artifactId>common-parent</artifactId>
    <version>6.0.0</version>
</parent>

の通り、Confluent Platformのバージョンとしては、6系である。

Confluent Platform and Apache Kafka Compatibility によると、 Confluent Platform 6系のKafkaバージョンは2.6.Xである。

io.confluent.kafka.connect.datagen.DatagenConnector

io.confluent.kafka.connect.datagen.DatagenConnector クラスは、 org.apache.kafka.connect.source.SourceConnector を継承している。割と素直な実装。

io.confluent.kafka.connect.datagen.DatagenConnector#start メソッドは特別なことはしておらず、コンフィグをロードするだけ。

io.confluent.kafka.connect.datagen.DatagenConnector#taskConfigs メソッドも特別なことはしていない。start時に受け取ったプロパティから taskConfigを生成して返す。

io.confluent.kafka.connect.datagen.DatagenConnector#stop メソッドおよび、 io.confluent.kafka.connect.datagen.DatagenConnector#config もほぼ何もしない。

タスクには io.confluent.kafka.connect.datagen.DatagenTask クラスを利用する。

io.confluent.kafka.connect.datagen.DatagenTask クラス

io.confluent.kafka.connect.datagen.DatagenTask#start メソッドが overrideされている。以下、ポイントを確認する。

オフセット管理の仕組みあり。

io/confluent/kafka/connect/datagen/DatagenTask.java:133

Map<String, Object> offset = context.offsetStorageReader().offset(sourcePartition);
if (offset != null) {
  //  The offset as it is stored contains our next state, so restore it as-is.
  taskGeneration = ((Long) offset.get(TASK_GENERATION)).intValue();
  count = ((Long) offset.get(CURRENT_ITERATION));
  random.setSeed((Long) offset.get(RANDOM_SEED));
}

io.confluent.avro.random.generator.Generator のジェネレータ（のビルダ）を利用する。

io/confluent/kafka/connect/datagen/DatagenTask.java:141

1
2
3

Generator.Builder generatorBuilder = new Generator.Builder()
    .random(random)
    .generation(count);

クイックスタートの設定があれば、それに従ってスキーマを読み込む。

io/confluent/kafka/connect/datagen/DatagenTask.java:144

String quickstartName = config.getQuickstart();
if (quickstartName != "") {
  try {
    quickstart = Quickstart.valueOf(quickstartName.toUpperCase());
    if (quickstart != null) {
      schemaFilename = quickstart.getSchemaFilename();
      schemaKeyField = schemaKeyField.equals("")
          ? quickstart.getSchemaKeyField() : schemaKeyField;
      try {
        generator = generatorBuilder
            .schemaStream(getClass().getClassLoader().getResourceAsStream(schemaFilename))
            .build();
      } catch (IOException e) {
        throw new ConnectException("Unable to read the '"
            + schemaFilename + "' schema file", e);
      }
    }
  } catch (IllegalArgumentException e) {
    log.warn("Quickstart '{}' not found: ", quickstartName, e);
  }

指定されたクイックスタート名に従い、パッケージに含まれるスキーマファイルを読み込み、それを適用しながらジェネレータを生成する。

なお、クイックスタートのたぐいはenumで定義されている。

io/confluent/kafka/connect/datagen/DatagenTask.java:75

protected enum Quickstart {
  CLICKSTREAM_CODES("clickstream_codes_schema.avro", "code"),
  CLICKSTREAM("clickstream_schema.avro", "ip"),
  CLICKSTREAM_USERS("clickstream_users_schema.avro", "user_id"),
  ORDERS("orders_schema.avro", "orderid"),
  RATINGS("ratings_schema.avro", "rating_id"),
  USERS("users_schema.avro", "userid"),
  USERS_("users_array_map_schema.avro", "userid"),
  PAGEVIEWS("pageviews_schema.avro", "viewtime"),
  STOCK_TRADES("stock_trades_schema.avro", "symbol"),
  INVENTORY("inventory.avro", "id"),
  PRODUCT("product.avro", "id");

  private final String schemaFilename;
  private final String keyName;

  Quickstart(String schemaFilename, String keyName) {
    this.schemaFilename = schemaFilename;
    this.keyName = keyName;
  }

  public String getSchemaFilename() {
    return schemaFilename;
  }

  public String getSchemaKeyField() {
    return keyName;
  }
}

クイックスタートが設定されておらず、スキーマの文字列が与えられた場合は、それを用いてジェネレータが生成される。

io/confluent/kafka/connect/datagen/DatagenTask.java:164

1 2	} else if (schemaString != "") { generator = generatorBuilder.schemaString(schemaString).build();

それ以外の場合、つまりスキーマ定義の書かれたファイルを指定する場合は、以下の通り。

io/confluent/kafka/connect/datagen/DatagenTask.java:166

{
      String err = "Unable to read the '" + schemaFilename + "' schema file";
      try {
        generator = generatorBuilder.schemaStream(new FileInputStream(schemaFilename)).build();
      } catch (FileNotFoundException e) {
        // also look in jars on the classpath
        try {
          generator = generatorBuilder
              .schemaStream(DatagenTask.class.getClassLoader().getResourceAsStream(schemaFilename))
              .build();
        } catch (IOException inner) {
          throw new ConnectException(err, e);
        }
      } catch (IOException e) {
        throw new ConnectException(err, e);
      }
    }

最後のAvroに関連する情報を生成して終了。

io/confluent/kafka/connect/datagen/DatagenTask.java:184

1
2
3

avroSchema = generator.schema();
avroData = new AvroData(1);
ksqlSchema = avroData.toConnectSchema(avroSchema);

io.confluent.kafka.connect.datagen.DatagenTask#poll メソッドもoverrideされている。以下、ポイントを記載する。

インターバル機能あり。

io/confluent/kafka/connect/datagen/DatagenTask.java:192

if (maxInterval > 0) {
  try {
    Thread.sleep((long) (maxInterval * Math.random()));
  } catch (InterruptedException e) {
    Thread.interrupted();
    return null;
  }
}

ジェネレータを利用し、オブジェクトを生成する。

io/confluent/kafka/connect/datagen/DatagenTask.java:201

final Object generatedObject = generator.generate();
if (!(generatedObject instanceof GenericRecord)) {
  throw new RuntimeException(String.format(
      "Expected Avro Random Generator to return instance of GenericRecord, found %s instead",
      generatedObject.getClass().getName()
  ));
}
final GenericRecord randomAvroMessage = (GenericRecord) generatedObject;

生成されたオブジェクトから、スキーマ定義に基づいてフィールドの値を取り出し、バリューのArrayListを生成する。

io/confluent/kafka/connect/datagen/DatagenTask.java:210

final List<Object> genericRowValues = new ArrayList<>();
for (org.apache.avro.Schema.Field field : avroSchema.getFields()) {
  final Object value = randomAvroMessage.get(field.name());
  if (value instanceof Record) {
    final Record record = (Record) value;
    final Object ksqlValue = avroData.toConnectData(record.getSchema(), record).value();
    Object optionValue = getOptionalValue(ksqlSchema.field(field.name()).schema(), ksqlValue);
    genericRowValues.add(optionValue);
  } else {
    genericRowValues.add(value);
  }
}

キーも同様に取り出し、Kafka Connectの形式に変換する。

io/confluent/kafka/connect/datagen/DatagenTask.java:224

SchemaAndValue key = new SchemaAndValue(DEFAULT_KEY_SCHEMA, null);
if (!schemaKeyField.isEmpty()) {
  key = avroData.toConnectData(
      randomAvroMessage.getSchema().getField(schemaKeyField).schema(),
      randomAvroMessage.get(schemaKeyField)
  );
}

先程ArrayListとして取り出したバリューもKafka Connect形式に変換する。

io/confluent/kafka/connect/datagen/DatagenTask.java:233

1 2	final org.apache.kafka.connect.data.Schema messageSchema = avroData.toConnectSchema(avroSchema); final Object messageValue = avroData.toConnectData(avroSchema, randomAvroMessage).value();

イテレートのたびに、メタデータを更新する。

io/confluent/kafka/connect/datagen/DatagenTask.java:246

// The source offsets will be the values that the next task lifetime will restore from
// Essentially, the "next" state of the connector after this loop completes
Map<String, Object> sourceOffset = new HashMap<>();
// The next lifetime will be a member of the next generation.
sourceOffset.put(TASK_GENERATION, taskGeneration + 1);
// We will have produced this record
sourceOffset.put(CURRENT_ITERATION, count + 1);
// This is the seed that we just re-seeded for our own next iteration.
sourceOffset.put(RANDOM_SEED, seed);

最後に、SourceRecordのリスト形式に変換し、レコードとして生成して戻り値として返す。

io/confluent/kafka/connect/datagen/DatagenTask.java:261

final List<SourceRecord> records = new ArrayList<>();
SourceRecord record = new SourceRecord(
    sourcePartition,
    sourceOffset,
    topic,
    null,
    key.schema(),
    key.value(),
    messageSchema,
    messageValue,
    null,
    headers
);
records.add(record);
count += records.size();
return records;

つづいて、 io.confluent.kafka.connect.datagen.DatagenTask#stop メソッドだが、これは特に何もしない。

io.confluent.kafka.connect.datagen.DatagenTask#getOptionalSchema というオプショナルなフィールドに関するスキーマを取得するためのヘルパーメソッドもある。 io.confluent.kafka.connect.datagen.DatagenTask#getOptionalValue メソッドもある。

動作確認

confluentinc-kafka-connect-datagen-0.4.0.zip をダウンロードし、 /opt/connectors以下に展開したものとする。

今回は以下の設定ファイルを参考に、データ生成してみる。なお、イテレーション回数は適度に修正して用いることを推奨する。

confluentinc-kafka-connect-datagen-0.4.0/etc/connector_users.config

{
  "name": "datagen-users",
  "config": {
    "connector.class": "io.confluent.kafka.connect.datagen.DatagenConnector",
    "kafka.topic": "users",
    "quickstart": "users",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "max.interval": 1000,
    "iterations": 10000000,
    "tasks.max": "1"
  }
}

上記はconfluentコマンド用で利用する際のコンフィグファイルである。そこで以下のようなKafka Connect用の設定ファイルを生成する。

/opt/connectors/confluentinc-kafka-connect-datagen-0.4.0/etc/connector_users.properties

name=users
connector.class=io.confluent.kafka.connect.datagen.DatagenConnector
kafka.topic=users
quickstart=users
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
max.interval=1000
iterations=10
tasks.max=1

スタンドアローンモードでKafka Connectを起動する。

1
2
3

$ sudo -u kafka /opt/kafka_pseudo/default/bin/connect-standalone.sh \
  /opt/kafka_pseudo/default/config/connect-standalone.properties \
  /opt/connectors/confluentinc-kafka-connect-datagen-0.4.0/etc/connector_users.properties

停止した後、結果を確認する。トピックが作られたことがわかる。

$ /opt/kafka_pseudo/default/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
__consumer_offsets
_confluent-command
syslog
users

データを確認する。

$ /opt/kafka_pseudo/default/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic users --from-beginning
{"registertime":1501850210149,"userid":"User_8","regionid":"Region_3","gender":"FEMALE"}
{"registertime":1516539876299,"userid":"User_2","regionid":"Region_7","gender":"OTHER"}
{"registertime":1505292095234,"userid":"User_4","regionid":"Region_1","gender":"OTHER"}
{"registertime":1502118362741,"userid":"User_3","regionid":"Region_1","gender":"FEMALE"}
{"registertime":1503193759324,"userid":"User_9","regionid":"Region_5","gender":"MALE"}
{"registertime":1507693509191,"userid":"User_1","regionid":"Region_8","gender":"OTHER"}
{"registertime":1497764008309,"userid":"User_1","regionid":"Region_6","gender":"OTHER"}
{"registertime":1514606256206,"userid":"User_1","regionid":"Region_3","gender":"MALE"}
{"registertime":1492595638722,"userid":"User_2","regionid":"Region_6","gender":"MALE"}
{"registertime":1500602208014,"userid":"User_3","regionid":"Region_1","gender":"MALE"}

ダミーデータが生成されていることが確認できた。

（WIP）

Create projects includes sbt launcher

2020-12-07

ScalaSBT

SBT, Scala

参考
メモ

参考

メモ

プロジェクト内に、sbt自体を含めてクローンしただけでビルドできるようにしたい、という動機。現時点ではCoursierプロジェクトが汎用性※が高く便利であると、個人的には感じた。

※ここでは、多くの環境で実行可能で、様々なツールを一度にセットアップ可能という意味。

Coursierプロジェクト

Coursier を使って最速でScalaの開発環境を整えるのブログに記載されているとおり、 Scala開発環境を簡単に整えられる。

ひとまずDocker内で試す。

1
2
3

$ sudo docker pull ubuntu:18.04
$ sudo docker run -rm -it -d --name ubu ubuntu:18.04
$ sudo docker exec -it ubu /bin/bash

Dockerコンテナ内でインストール。

# apt update
# apt install -y curl
# curl -fLo cs https://git.io/coursier-cli-linux &&
     chmod +x cs &&
     ./cs

結果として以下がインストールされた様子。

Checking if the standard Scala applications are installed
  Installed ammonite
  Installed cs
  Installed coursier
  Installed scala
  Installed scalac
  Installed sbt
  Installed sbtn
  Installed scalafmt

~/.profileに環境変数等がまとまっているので有効化する。

1	# source ~/.profile

これでインストールされたコマンドが使用できるようになった。

なお、参考までに上記でダウンロードしたcsコマンドは以下の通り。

$ ./cs setup --help
Command: setup
Usage: cs setup
  --jvm  <string?>
  --jvm-dir  <string?>
  --system-jvm  <bool?>
  --local-only  <bool>
  --update  <bool>
  --jvm-index  <string?>
  --graalvm-home  <string?>
  --graalvm-option  <string*>
  --graalvm-default-version  <string?>
  --install-dir | --dir  <string?>
  --install-platform  <string?>
        Platform for prebuilt binaries (e.g. "x86_64-pc-linux", "x86_64-apple-darwin", "x86_64-pc-win32")
  --install-prefer-prebuilt  <bool>
  --only-prebuilt  <bool>
        Require prebuilt artifacts for native applications, don't try to build native executable ourselves
  --repository | -r  <maven|sonatype:$repo|ivy2local|bintray:$org/$repo|bintray-ivy:$org/$repo|typesafe:ivy-$repo|typesafe:$repo|sbt-plugin:$repo|ivy:$pattern>
        Repository - for multiple repositories, separate with comma and/or add this option multiple times (e.g. -r central,ivy2local -r sonatype:snapshots, or equivalently -r central,ivy2local,sonatype:snapshots)
  --default-repositories  <bool>
  --proguarded  <bool?>
  --channel  <org:name>
        Channel for apps
  --default-channels  <bool>
        Add default channels
  --contrib  <bool>
        Add contrib channel
  --file-channels  <bool>
        Add channels read from the configuration directory
  --cache  <string?>
        Cache directory (defaults to environment variable COURSIER_CACHE, or ~/.cache/coursier/v1 on Linux and ~/Library/Caches/Coursier/v1 on Mac)
  --mode | -m  <offline|update-changing|update|missing|force>
        Download mode (default: missing, that is fetch things missing from cache)
  --ttl | -l  <duration>
        TTL duration (e.g. "24 hours")
  --parallel | -n  <int>
        Maximum number of parallel downloads (default: 6)
  --checksum  <checksum1,checksum2,...>
        Checksum types to check - end with none to allow for no checksum validation if no checksum is available, example: SHA-256,SHA-1,none
  --retry-count  <int>
        Retry limit for Checksum error when fetching a file
  --cache-file-artifacts | --cfa  <bool>
        Flag that specifies if a local artifact should be cached.
  --follow-http-to-https-redirect  <bool>
        Whether to follow http to https redirections
  --credentials  <host(realm) user:pass|host user:pass>
        Credentials to be used when fetching metadata or artifacts. Specify multiple times to pass multiple credentials. Alternatively, use the COURSIER_CREDENTIALS environment variable
  --credential-file  <string*>
        Path to credential files to read credentials from
  --use-env-credentials  <bool>
        Whether to read credentials from COURSIER_CREDENTIALS (env) or coursier.credentials (Java property), along those passed with --credentials and --credential-file
  --quiet | -q  <counter>
        Quiet output
  --verbose | -v  <counter>
        Increase verbosity (specify several times to increase more)
  --progress | -P  <bool>
        Force display of progress bars
  --env  <bool>
  --user-home  <string?>
  --banner  <bool?>
  --yes | -y  <bool?>
  --try-revert  <bool>
  --apps  <string*>

（補足）sbt-launcher-packageプロジェクト

sbt-launcher-package をビルドすることでSBTランチャーを提供できそうだった。

（補足）launcherプロジェクト

https://github.com/sbt/launcher/blob/1.x/.java-version にも記載されているとおり、 JDK7系にしか公式に対応していなかった。