社内se × プログラマ × ビッグデータ

プログラミングなどITに興味があります。

タイガースとジャイアンツが含まれるツイート件数をカウントしてみた

Spark
データ取得期間

2018/11/11 - 11/13 内の数時間

データ取得方法
  • Twitter API（検索） を叩くバッチを1分間隔で実行。
  • 検索条件のキーワードとして、”タイガース” もしくは、”ジャイアンツ”が含まれること。
  • リツイートは取得対象外
  • たまに単に球団名が列挙されているツイートが存在するが、特に除外はしていない（取得対象内）
集計方法

Apache spark の spark-shell を使う。
とは言っても、特別な分析ライブラリを使用するわけでもなく、単にRDDとして取り込んで極めて基本的な関数を呼び出すだけ。

集計ログ

取得した総ツイート件数

scala> val rawData = sc.textFile("タイガース_ジャイアンツ.txt")
rawData: org.apache.spark.rdd.RDD[String] = タイガース_ジャイアンツ.txt MapPartitionsRDD[19] at textFile at <console>:27

scala> rawData.count
res24: Long = 1650

ツイートにタイガースが含まれる件数

scala> def isTigers(line: String): Boolean = line.contains("タイガース")
isTigers: (line: String)Boolean

scala> rawData.filter(isTigers).count
res25: Long = 551

ツイートにジャイアンツが含まれる件数

scala> def isGiants(line: String): Boolean = line.contains("ジャイアンツ")
isGiants: (line: String)Boolean

scala> rawData.filter(isGiants).count
res26: Long = 1196

1つのツイートにタイガースとジャイアンツの両方が含まれているものもある。
しかし、ジャイアンツが含まれているツイート数がタイガースよりも、およそ2倍という結果に。

ツイートに矢野監督が含まれる件数

scala> def isYano(line: String) = line.contains("矢野監督")
isYano: (line: String)Boolean

scala> rawData.filter(isYano).count
res30: Long = 5

ツイートに原監督が含まれる件数

scala> def isHara(line: String) = line.contains("原監督")
isHara: (line: String)Boolean

scala> rawData.filter(isHara).count
res31: Long = 11

もちろん、これだけで世間の注目度を示しているわけではないし、集計期間や集計方法にも問題がたくさんあると思いますが、それにしても思っていた以上に数値に差が出る結果となりました。
もう少しサンプルデータを増やした上で、別の角度からも集計をしてみたい。

grep -E が正規表現対応オプションであると勘違いしていた

Linux

grep って、global regular expression print の略なので、それ自体が正規表現で一致したものを抽出するコマンドなんですね。
正規表現を使いたい時に、-E オプションを付けるものだと勘違いしていました。

いま、以下のように2行が書かれたファイル（test.txt）があるとします。

$ cat > test.txt
1024
hatena
$ grep ^1 test.txt 
1024

はい。特に -E オプションを付けなくても正規表現でマッチしてくれました。

ところが。

$ grep '^(1|h)' test.txt

これは何も返してくれません。

そこで、-E オプションを付けてみると。

$ grep -E '^(1|h)' test.txt 
1024
hatena

期待する結果を返してくれました。
E オプションをつけることで、拡張正規表現と呼ばれる式が使用出来るようになります。

egrep でもOKです。

$ egrep '^(1|h)' test.txt 
1024
hatena

拡張正規表現を使える = 正規表現を使えると間違って覚えてしまっていたようです。
何気に使用しているオプションについて、意味を正しく理解していないのは宜しくないですね。

開発者が知っておくべき Couchbase についての 10項目

Couchbase
はじめに

公式ブログによって、こうやってまとめておいてくれると、読みやすいし初学者にとって助かります。
blog.couchbase.com

第10位
Document access in Couchbase is strongly consistent, query access is eventually consistent

Couchbase のデータには強い一貫性があると主張されています。
key / value アクセスなので、そこは保証されやすいところなのかと思います。
View （恐らくインデックス）についても、最終的には一貫性が保たれるということですが、逆に言うと一貫性がない瞬間（インデックスが生成されるまで）もあるということですね。

第9位
Writes are asynchronous by default but can be controlled

書き込み処理は基本的に非同期で行われるようです。
レプリカの生成と、データの永続化（メモリからディスクへの書き込み）はバックグラウンドで行われ、クライアント側はその結果の通知を受け取れるようです。
クライアント側で先に通知を受け取るか、通知よりも先に非同期にレプリカを作成させるかなどはクライアント側で選べるようです。

第8位
Couchbase has atomic operations for counting and appending

カウント処理と追加処理において、不可分操作をサポートしているようです。
例にあるように、incr は書き込み処理と結果を返す処理を行います。
要は追加に失敗したのに、追加されたものとしてカウントされてしまったりすることは無いということかと思います。

cb.set(“mykey”, 1)
x = cb.incr(“mykey”)
puts x #=> 2

第7位
Start with everything in one bucket

Bucket というものがデータベースのようなものであり、RDBMSでいうところテーブルというわけではない。
そのため、RDBMSからそのままデータを移すとしたら、複数のテーブルが１つのBucket に全て投入されることになります。
ただし、"typte" という属性が恐らくデータ毎に設定が可能であるため、それを設定することで同じBucket内にあるデータであっても差別化ができるようです。
とにかく１つのBucketでスタートすることが推奨されているようですね。

第6位
Try to use 5 or less buckets in Couchbase. Never more than 10.

データは固定のスキーマを持つわけではないので、色々なスキーマのデータを同じBucketに入れることが可能。
ソフトウェアとして限界値は定められていないものの、10 Bucketにもなると、CPUやDiskIOに問題が発生することが確認されているようです。
この辺りは Couchbase のバージョンアップによって、改善される可能性はあるかもしれませんが、出来る限り少ない数の Bucket で運用する方が良さそうですね。
まぁ、もしどうしても Bucket をたくさん作って管理を分けたいのであれば、別のクラスタに切り離してしまった方が良いのかもしれません。

第5位
Use CAS over GetL almost always

CAS は "Check and Set"の略のようです。
KVSにおけるトランザクション処理を行うために必要な操作のようです。
要は楽観的ロックと悲観的ロックの話をしていますが、基本的に楽観的ロックを使用するべきだということでしょうか。
Couchbaseにおけるロック機能というものを知らないので、これ以上は理解が出来ていません。

第4位
Use multi-get operations

Couchbase には、キーのリストから複数のレコードを同時に？検索できるようです。
個々のキーを一つずつ検索するよりも、高パフォーマンスが期待できるようです。

第3位
Keep your client libraries up-to-date

これは、Couchbase に特化した話でもないと思いますが、使用するライブラリは最新のものが良いよということですね。
バージョンアップが頻繁に行われている製品であれば、何であっても同じことが言えると思います。

第2位
Model your data using JSON documents

格納するデータはJSON形式で。
CouchbaseはJSON形式や、バイナリ形式のドキュメントをサポートしていますが、まずはJSON形式で試すことを推奨しているようです。
JSON形式で保存しておくと、インデックスが作成できたり、特殊な？クエリを投げられるなどメリットがありそうです。

第1位
Use indexes effectively

インデックスを効果的に使うこと。
出来る限り、プライマリキーでのアクセスを行うこと。キーとメタデータをメモリ上に持っているので、アクセスは速いはず。
セカンダリ・インデックスでのアクセスは、パフォーマンスを必要としない分析用などで使用するべきとのこと。

この時点で、セカンダリ・インデックスのパフォーマンスは、それ程期待しない方が良いのかもしれません。
4つの design documents と、1つの design documents あたり、view は10個以下に抑えること。
それでも多いようなきもしますが。
インデックスデータに対して、何の "reduce" 処理を行う必要がないのであれば、value に "null" を設定しておきべきとのこと。

おわりに

これを読んだだけでは表面的な部分しか分かりませんが、使用していく内に「あれはそういう意味だったのか」のように、それぞれの意味をより深く知ることが出来るのかもしれません。
”Use indexes effectively” というのが一見当たり前のように見えて、Couchbase を使う上で実はとても大切なことのような気がします。

新しく使用する技術に対して、「取り合えず使ってみる」アプローチと、「概念をまず理解する」アプローチどちらが良いかは人それぞれかと思います。
自分の場合、取り合えず使ってみて、分からないところが出てきたら調べることが多いように思います。
ただ、エラーを回避することに精一杯になってしまって、結局その製品のことをあまり理解できていない事に後で気づくので、手を少し止めてインプットする時間も意識して作れるといいなと思います。

リファクタリングと追加実装はコミットを分けてほしい

プログラミング

主にソースコードをレビューする立場である場合の視点になります。
チームで開発しているとき、リファクタリングと追加実装を同時にレビュー提出されることがあります。

ただ、レビュー依頼のコメントには、リファクタリングのことは触れられていないので、レビューする側としては、全てが追加機能の実装に伴う変更点であると最初は解釈します。

しかしながら、レビューを進めていくとどうも追加実装の要件とは関係の無いところまで編集されている様子。
よくよく確認してみると、気になった箇所のリファクタリングをついでにやったとのこと。

気になったソースコードをそのまま放置せずにリファクタリングする試みはとても良いことだと思うのですが、個人的には出来ればコミットは分けておきたいところです。
コミットを分けておくことで、リファクタリング箇所と追加実装箇所を明確に分けて見ることができるため、レビューもしやすくなります。

あと、リファクタリングを行った段階で単体テストを通しておけば、そのリファクタリングによって不具合が作られなかったことの確認も容易にできます。
以前あったのが、リファクタリングと追加実装を同時に行った後、単体テストが通らなくなり、ずっと追加実装したコードを疑っていたが、実はリファクタリングした箇所が不具合を作ってしまっていました。

チームで開発作業していると、他人が実装したプログラムを引き継いだり、逆に自分が実装したプログラムを引き継いてもらったりなど日常茶飯事なので、少しずつ小さなステップで確認しながら実装していかないと、簡単に不具合が混入されてしまいます。

１．コード規約が定まっていない（個人的にはあまり細かく縛られたくないですが）
２．ソースコード内のコメントが不親切
３．詳細設計書がない

こういった場合、良くも悪くもプログラマーの個性がソースコードに表れるため、理解が難しくなる場合があります。
※数ヵ月後には自分自身が書いたものですら、理解出来なくなってしまうことも

切羽詰っているプロジェクトほど、こういったステップを踏む余裕はないかもしれませんが、後戻りを防ぐためにも確実に進めるようにしていきたいものです。

Java 無名（匿名）クラスを意識して使ってみる

java

Javaでプログラムの基本は、クラスのインスタンスを生成し、そのインスタンスから目的のメソッドを呼ぶところにあると思いますが、結果的に一度しかインスタンス化されないクラス（オブジェクト）があったり、ある特定のクラスからしか必要とされないインスタンスがあったりします。

自分はプログラミングする際、1つのクラスに多くの役割を持たせたくない（役割分担させる）ため、クラスを細かく定義することを基本にしていますが、中には一度しかインスタンスが生成されない、メソッドが一つしかないクラスを作ってしまうことがあります。

特に大きな問題になる訳ではないですが、単純な処理しか行っていないのであれば、そのメソッドを定義する無理にクラスを定義せずに、その処理を必要としているクラス内に含めてしまっても良いかもしれません。

無名クラスという仕組みを使えば、インターフェースを実装したクラス（あるいはあるクラスを継承したサブクラス）の宣言を省略することが出来るので、その辺りを上手く使えないかと試みたのですが（今までは意識して使ったことなし）。

クラスを定義しインスタンスを明示的に生成する場合

interface Dao {
    String getData();
}
	
class MySqlDao implements Dao {
    @Override
    public String getData() {
	return "MySql data from MySqlDao.class";
    }
}

MySqlDao mySqlDao = new MySqlDao();
System.out.println(mySqlDao.getData());

Dao インターフェースを実装した MySqlDao クラスを定義しています。
呼び出し側では、そのインスタンスを生成しています。

無名クラスを使う場合

interface Dao {
    String getData();
}

Dao mySqlDao2 = new Dao() {
    @Override
    public String getData() {
	return "MySql data from anonymous class";
    }
};
System.out.println(mySqlDao2.getData());

まるで Dao のインスタンスを生成しているように見えますが、ここでは Dao インターフェースを実装した無名クラスのインスタンス(mySqlDao2 )が生成されていることになります。
※そもそもインターフェースのインスタンスを生成することはできない

結局、クラスの宣言と利用を同時に行っているだけではあるのですが、これによって MySqlDao.class の定義を省略することが可能になりました。
管理するクラスの数が減るという面でメリットがあると思います。

一方で”無名”のクラスになってしまうので、そのクラス名からどのような役割を持っているのかを推測することが出来なくなります。
例が悪かったですが、Dao などは外部から呼び出される前提のクラスになると思いますので、広く使われるようなクラスは無名クラスにするべきではなさそうです。
本当にそこでしか使わない（使い捨てに出来る）クラス限定で適用するものかと思います。
さらに無名クラスになれるのは、インターフェースを実装したクラスあるいはサブクラスに限定されるので、今のところ個人的には活躍の場はそれ程ない見込みです。

Java8 日時APIがわからない（現在時刻の取得編）

java

もうすぐ、JDK8(java8)の商用サポート期限が2019年1月で終了するという中、未だに Java8 日時API に混乱させられています。。
※この手の情報は色んなところに既にまとめられていますし、自分は5年くらい遅れている気がしますが、自分の備忘録のために覚えたことを残しておきたいと思います

実行マシンの timezone は JST です。

$ date
2018年  9月 28日 金曜日 00:46:31 JST

どんなクラスがあるか

クラス 内容
Instant 日時（エポック秒）
LocalDateTime 日付時刻（タイムゾーンなし）
ZonedDateTime タイムゾーンつきの日付時刻
OffsetDateTime オフセット付きの日付時刻

now() で現在時刻を取得してみる

Instant instantNow = Instant.now();
System.out.println("instantNow:" + instantNow);

LocalDateTime ldtNow = LocalDateTime.now();
System.out.println("ldtNow:" + ldtNow);

ZonedDateTime zdtNow = ZonedDateTime.now();
System.out.println("zdtNow:" + zdtNow);

OffsetDateTime odtNow = OffsetDateTime.now();
System.out.println("odtNow:" + odtNow);

結果

Instant.now(): 2018-09-27T16:05:38.683Z
LocalDateTime.now(): 2018-09-28T01:05:39.077
ZonedDateTime.now(): 2018-09-28T01:05:39.078+09:00[Asia/Tokyo]
OffsetDateTime.now(): 2018-09-28T01:05:39.079+09:00

Instant.now() で取得した時刻は、マイナス9時間されている、つまりUTCで取得されていることが解ります。
ZonedDateTime と OffsetDateTime に関しても、9時間の差分が UTC に対してあることが明示されています。

タイムゾーンIDを指定してみる

LocalDateTime ldtjst = LocalDateTime.now(ZoneId.of("Asia/Tokyo"));
System.out.println("LocalDateTime.now(): " + ldtjst);

ZonedDateTime zdtjst = ZonedDateTime.now(ZoneId.of("Asia/Tokyo"));
System.out.println("ZonedDateTime.now(): " + zdtjst);

OffsetDateTime odtjst = OffsetDateTime.now(ZoneId.of("Asia/Tokyo"));
System.out.println("OffsetDateTime.now(): " + odtjst);

結果

LocalDateTime.now(): 2018-09-28T01:13:47.681
ZonedDateTime.now(): 2018-09-28T01:13:47.681+09:00[Asia/Tokyo]
OffsetDateTime.now(): 2018-09-28T01:13:47.681+09:00

※Instant.now() には、ZoneId を指定できません

結果、見た目は特に指定しなかった場合と変わらず。これは指定しなかった場合も、JST として解釈されているためと思われます。

タイムゾーンを変換してみる(JST -> UTC)

LocalDateTime ldtjst = LocalDateTime.now(ZoneId.of("Asia/Tokyo"));
LocalDateTime ldtutc = LocalDateTime.ofInstant(ldtjst.toInstant(ZoneOffset.UTC), ZoneId.of("UTC"));
System.out.println("LocalDateTime.now(): " + ldtutc);

ZonedDateTime zdtjst = ZonedDateTime.now(ZoneId.of("Asia/Tokyo"));
ZonedDateTime zdtutc = ZonedDateTime.ofInstant(zdtjst.toInstant(), ZoneId.of("UTC"));
System.out.println("ZonedDateTime.now(): " + zdtutc);

OffsetDateTime odtjst = OffsetDateTime.now(ZoneId.of("Asia/Tokyo"));
OffsetDateTime odtutc = OffsetDateTime.ofInstant(odtjst.toInstant(), ZoneId.of("UTC"));
System.out.println("OffsetDateTime.now(): " + odtutc);

結果

LocalDateTime.now(): 2018-09-28T01:25:50.328
ZonedDateTime.now(): 2018-09-27T16:25:50.329Z[UTC]
OffsetDateTime.now(): 2018-09-27T16:25:50.329Z

LocalDateTime.now() の見た目には変化なし、これはそもそも LocalDateTime がタイムゾーンを持たないから。
ZonedDateTime と OffsetDateTime は、マイナス9時間された UTCで取得されていることが解ります。
時刻の最後に "Z" が付いていますが、これは Zero timezone、つまり UTC からのオフセットが 0 であるという意味になります。

所感
現在時刻を取得しただけですが、各クラスがどのような意味を持つのかや、タイムゾーンを指定、変換したときの振る舞いを確認することが出来ました。
ZonedDateTime と OffsetDateTime の違いが解りにくいことと、各クラス間での変換については調べられていませんが、これだけでも、自分の中では少し整理できたように思います。

追記
"yyyyMMddHHmmss" は西暦を元にフォーマットしてくれる仕様。
"YYYYMMddHHmmss" はその年の最初の木曜日が含まれる週はたとえ昨年の日付（12/31）であっても今年としてフォーマットする仕様。

String ORG_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'";
String orgDate = "2017-12-31T13:00:01.247Z";
LocalDateTime ldt = LocalDateTime.parse(orgDate, DateTimeFormatter.ofPattern(ORG_FORMAT));
		
// yyyyMMddHHmmss
String NEW_FORMAT = "yyyyMMddHHmmss";
String result = ldt.format(DateTimeFormatter.ofPattern(NEW_FORMAT));
System.out.println(result); // 20171231130001

// YYYYMMddHHmmss
String NEW_FORMAT2 = "YYYYMMddHHmmss";
String result2 = ldt.format(DateTimeFormatter.ofPattern(NEW_FORMAT2));
System.out.println(result2); // 20181231130001

Apache Spark2.3 ブロードキャスト変数のパフォーマンス

Spark java

ブロードキャスト変数は、リードオンリーの変数を効率的に各 Executor に送信する仕組みです。

Apache Spark2 にて、ブロードキャスト変数のパフォーマンスをローカル環境で確認してみました。
スレッド数は 3 を指定しています。
※ sparkConf.setMaster("local[3]");

ブロードキャスト変数なしの場合
1. 100 万件の数値リストを生成
2. 3件の文字列リストから RDD を生成
3. RDD の各文字列要素に対して、100 万件の数値リストをループし付加する処理(100万回ループ)を行う。
※ 即ち、3 の処理が (spark)Executor で行われる
計測結果(sec)：
00:00:07.572
00:00:09.765
00:00:10.879

ブロードキャスト変数ありの場合
1. 100 万件の数値リストを生成し、それをブロードキャスト変数にする
2. 3件の文字列リストから RDD を生成
3. RDD の各文字列要素に対して、100 万件の数値リストをブロードキャスト変数で渡し、ループし付加する処理(100万回ループ)を行う。
※ 即ち、3 の処理が (spark)Executor で行われる
計測結果(sec)：
00:00:01.593
00:00:01.802
00:00:01.668

ブロードキャスト変数を用いて、100万件の数値リストを Executor に渡した方が、圧倒的に速いという結果になりました。
いくら何でも違いが大きい印象を受けたので、これは Spark2 の恩恵かも？しれないと思い、Spark1.6でも試してみたところ
Spark1.6 ブロードキャスト変数なし(sec)：
0:00:09.843
0:00:08.355
0:00:08.403

Spark1.6 ブロードキャスト変数あり(sec)：
0:00:01.594
0:00:00.890
0:00:00.899

結果は誤差の範囲でした。むしろ1.6の方がほんの少し速いという結果に。
テストで使用したコード。
github.com