2014-12-29

Ratpack で Java Web アプリケーション作成

Java

Ratpack は以前「Ratpack + JHaml + Morphia で MongoDB を使った Web アプリ開発」で試しましたが、3年以上経っているので改めて試してみました。

今回は単純な Java Web アプリケーションを Ratpack で作成する事にします。

Ratpack 0.9.11

ソースは http://github.com/fits/try_samples/tree/master/blog/20141229/

Web アプリケーションの作成

今回の作成手順は以下のようになります。

(1) HandlerFactory インターフェースの実装クラスを作成
(2) ratpack.properties で handlerFactory を設定

ファイル構成

今回は Gradle でビルドしましたので、ファイルは以下のような構成になっています。

build.gradle
src/main/java/AppHandlerFactory.java
src/ratpack/ratpack.properties

ビルド定義

アプリケーションの実行やアーカイブ化を簡単に行うため、Gradle 用の ratpack-java プラグインを使います。

ratpack-java を使う場合、ratpack の依存設定（dependencies）は要らないようですが、slf4j だけは実行に要るようです。

build.gradle

buildscript {
    repositories {
        jcenter()
    }

    dependencies {
        classpath 'io.ratpack:ratpack-gradle:0.9.11'
    }
}

apply plugin: 'io.ratpack.ratpack-java'

repositories {
    jcenter()
}

dependencies {
    runtime 'org.slf4j:slf4j-simple:1.7.5'
}

(1) HandlerFactory 実装クラスの作成

とりあえず /sample/xxx へアクセスすると単に "sample - xxx" を出力するだけの処理を実装しました。（xxx は任意の文字列）

出力には Context オブジェクトの render() メソッドを使用し、URL のパラメータ部分（下記の :id）は PathTokens の get() メソッドで取得します。

src/main/java/AppHandlerFactory.java

import static ratpack.handling.Handlers.*;

import ratpack.handling.Context;
import ratpack.handling.Handler;
import ratpack.launch.HandlerFactory;
import ratpack.launch.LaunchConfig;

public class AppHandlerFactory implements HandlerFactory {
    @Override
    public Handler create(LaunchConfig config) throws Exception {
        return chain(
            path("sample/:id", ctx -> 
                ctx.render("sample - " + ctx.getPathTokens().get("id")))
        );
    }
}

(2) handlerFactory の設定

作成した AppHandlerFactory を handlerFactory へ設定します。

こうする事で、実行時に handlerFactory として AppHandlerFactory が使用されます。

src/ratpack/ratpack.properties

handlerFactory=AppHandlerFactory

テスト実行

gradle run でテスト実行できます。

> gradle run

:compileJava UP-TO-DATE
:processResources UP-TO-DATE
:classes UP-TO-DATE
:configureRun
:prepareBaseDir UP-TO-DATE
:run
[main] INFO ratpack.server.internal.NettyRatpackServer - Ratpack started for http://localhost:5050
・・・

http://localhost:5050/sample/abc へアクセスすると、sample - abc が表示されます。

成果物の生成

gradle distZip を実行すると、build/distributions へ zip ファイルが生成されます。

> gradle distZip

:compileJava UP-TO-DATE
:processResources UP-TO-DATE
:classes UP-TO-DATE
:jar
:startScripts
:distZip

BUILD SUCCESSFUL

zip ファイルを解凍して、bin ディレクトリの起動スクリプト（今回のサンプルでは ratpack_sample）を実行すれば Web アプリケーションを単体起動できます。

2014-12-21

MyBatis / iBatis の動的 SQL を API で作成

Java Groovy

MyBatis / iBatis の API を使って DB へ接続せずに Mapper XML の動的 SQL を作成する方法です。

ソースは http://github.com/fits/try_samples/tree/master/blog/20141221/

MyBatis の場合

動的 SQL の結果を取得する手順は下記のようになります。

(1) Configuration をインスタンス化
(2) (1) と Mapper XML で XMLMapperBuilder をインスタンス化
(3) Mapper XML をパース
(4) (1) から指定の SQL に対応した MappedStatement を取得
(5) (4) で取得した MappedStatement へパラメータを渡して動的 SQL を構築、結果の SQL を取得

今回は Groovy で実装してみました。

動的 SQL のパラメータはコマンドライン引数で JSON 文字列として指定するようにしています。

mybatis_sql_gen.groovy

@Grab('org.mybatis:mybatis:3.2.8')
import org.apache.ibatis.session.*
import org.apache.ibatis.builder.xml.*

import groovy.json.JsonSlurper

if (args.length < 3) {
    println '<mybatis mapper xml> <sql id> <json params>'
    return
}
// (1)
def config = new Configuration()
// (2)
def parser = new XMLMapperBuilder(new File(args[0]).newInputStream(), config, "", config.sqlFragments)
// (3)
parser.parse()
// (4)
def st = config.getMappedStatement(args[1])
// パラメータの作成（JSON 文字列から）
def params = new JsonSlurper().parseText args[2]
// (5)
def sql = st.getBoundSql(params).sql

println sql

実行

下記 Mapper XML を使って実行してみます。

mapper.xml

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="sample">
  <select id="findData">
    SELECT * FROM data
    WHERE
      title like #{title}
    <if test="author != null and author.name != null">
      AND author_name like #{author.name}
    </if>
    <if test="types">
      AND type in 
      <foreach item="type" collection="types" open="(" separator="," close=")">
        #{type}
      </foreach>
    </if>
  </select>
</mapper>

実行例1

まずはパラメータ無しの場合。{}

> groovy mybatis_sql_gen.groovy mapper.xml findData "{}"

SELECT * FROM data
    WHERE
      title like ?

実行例2

次に、author.name と types パラメータを指定した場合。{"author":{"name": 1}, "types": [1, 2, 3]}

なお、今回の動的 SQL ではパラメータの有無しか見ていませんので、値には適当な数値（1や [1, 2, 3]）を使っています。

> groovy mybatis_sql_gen.groovy gen.groovy mapper.xml findData "{\"author\":{\"name\": 1}, \"types\": [1, 2, 3]}"

SELECT * FROM data
    WHERE
      title like ?

      AND author_name like ?


      AND type in
       (
        ?
       ,
        ?
       ,
        ?
       )

iBatis の場合

iBatis の場合も、使用する API は異なりますが同じような手順で処理できます。

ibatis_sql_gen.groovy

@Grab('org.apache.ibatis:ibatis-sqlmap:2.3.4.726')
import com.ibatis.sqlmap.engine.builder.xml.*
import com.ibatis.sqlmap.engine.scope.*

import groovy.json.JsonSlurper

if (args.length < 3) {
    println '<ibatis mapper xml> <sql id> <json params>'
    return
}

def state = new XmlParserState()

def parser = new SqlMapParser(state)
parser.parse(new File(args[0]).newInputStream())

// SqlMapExecutorDelegate を取得
def dlg = state.config.delegate

def st = dlg.getMappedStatement(args[1])
def sql = st.sql

def scope = new StatementScope(new SessionScope())
scope.statement = st

// パラメータの作成（JSON 文字列から）
def params = new JsonSlurper().parseText args[2]

println sql.getSql(scope, params)

実行

下記 Mapper XML を使って同様に実行してみます。

mapper.xml

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE sqlMap PUBLIC "-//ibatis.apache.org//DTD SQL Map 2.0//EN" "http://ibatis.apache.org/dtd/sql-map-2.dtd">
<sqlMap namespace="sample">
  <select id="findData">
    SELECT * FROM data
    WHERE
      title like #title#
    <isNotNull property="author">
      <isNotNull property="author.name">
        AND author_name like #author.name#
      </isNotNull>
    </isNotNull>
    <isNotNull property="types">
      AND type in
      <iterate property="types" open="(" conjunction="," close=")">
        #types[]#
      </iterate>
    </isNotNull>
  </select>
</sqlMap>

実行例1

> groovy ibatis_sql_gen.groovy mapper.xml findData "{}"

     SELECT * FROM data     WHERE       title like ?

実行例2

> groovy ibatis_sql_gen.groovy mapper.xml findData "{\"author\":{\"name\": 1}, \"types\": [1, 2, 3]}"

     SELECT * FROM data     WHERE       title like ?                     AND author_name like ?                        AND type in       (         ?       ,     ?       ,         ?       )

2014-12-09

Sodium で関数型リアクティブプログラミング2 - skip・take 処理

Java FRP

前回に続き、Sodium を試してみます。

Sodium

今回は「RxJava で行単位のファイル処理 - Groovy, Java Lambda」で実装したものと同等の処理を Sodium を使って実装してみました。

ソースは http://github.com/fits/try_samples/tree/master/blog/20141209/

skip・take 処理

Sodium には、RxJava の際に使った skip （指定した処理数だけ無視する）や take （指定した処理数だけ取り出す）のようなメソッドが用意されていないようなので、Event と Behavior を組み合わせて実装してみる事にします。

Event クラスには gate メソッドがあり、このメソッドを使えば Behavior オブジェクトの値が true の場合だけイベントを発生するような Event オブジェクトを作成できます。

つまり、指定した数のイベントを受け取れば true と false が反転する Behavior オブジェクトを用意して gate メソッドへ与えてやれば skip や take の処理を実現した Event オブジェクトを作成できそうです。

なお、skip と take の違いは、false から true への変化か true から false への変化かの違いしかありませんので、下記のサンプルでは batch メソッドとして共通化し、判定処理を引数 Lambda1<Integer, Boolean> cond として渡すようにしました。

batch メソッドは以下のような処理内容となっています。

(1) イベントの発生数をカウントアップする Behavior を用意（実際は BehaviorSink を使用）
(2) イベント発生時に (1) をカウントアップ
(3) (1) のカウント値が条件に合致した場合のみイベントを発生させる Event を作成

ReadLineFile.java

import sodium.*;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.function.*;

class ReadLineFile {
    public static void main(String... args) throws Exception {
        // skip 処理
        Function<Integer, Function<Event<String>, Event<String>>> skip = n -> {
            return ev -> batch( v -> v >= n, n, ev);
        };
        // take 処理
        Function<Integer, Function<Event<String>, Event<String>>> take = n -> {
            return ev -> batch( v -> v < n, n, ev);
        };

        // 1行スキップして 3行取得する Event を作成する処理を合成 (b) (c)
        Function<Event<String>, Event<String>> skipAndTake3 = skip.apply(1).andThen( take.apply(3) );

        // (a)
        EventSink<String> es = new EventSink<>();
        // (d)
        Listener esl = skipAndTake3.apply(es).map( v -> "# " + v ).listen( System.out::println );
        // ファイルを行単位で処理
        readFileLines(args[0], es);

        esl.unlisten();
    }

    private static void readFileLines(String fileName, EventSink<String> es) throws IOException {
        // ファイルを行単位で EventSink へ send
        try (BufferedReader br = new BufferedReader(new FileReader(fileName))) {
            br.lines().forEach( es::send );
        }
    }
    // skip・take の共通処理
    private static Event<String> batch(Lambda1<Integer, Boolean> cond, int n, Event<String> ev) {
        // (1) イベント発生数をカウントする Behavior を用意
        BehaviorSink<Integer> counter = new BehaviorSink<>(0);
        // (2) イベント発生時に counter をカウントアップ
        ev.listen( v -> counter.send(counter.sample() + 1) );
        // (3) counter の値が条件に合致した場合のみイベント発生する Event を作成
        return ev.gate(counter.map(cond));
    }
}

上記は、以下の 4つの Event が (a) -> (b) -> (c) -> (d) の順でイベントを伝播するような構成となっており、(d) に到達した場合のみ System.out::println を実行するようになっています。

(a) EventSink
(b) skip で作成した Event
(c) take で作成した Event
(d) 先頭に "# " を付与する Event

また、batch メソッドの実装内容に関しては counter の無駄なカウントアップを防止するため、下記のように unlisten するようにした方が望ましいかもしれません。

unlisten する処理を追加した batch メソッドの例

    private static Event<String> batch(Lambda1<Integer, Boolean> cond, int n, Event<String> ev) {
        BehaviorSink<Integer> counter = new BehaviorSink<>(0);
        final ArrayList<Listener> list = new ArrayList<>(1);

        list.add(ev.listen( v -> {
            int newValue = counter.sample() + 1;
            counter.send(newValue);

            if (newValue >= n) {
                list.stream().forEach( li -> li.unlisten() );
            }
        }));

        return ev.gate(counter.map(cond));
    }

実行

それでは、下記ファイルを使って実行してみます。

test1.txt

1a
2b
3c
4d
5e

実行結果は以下の通りです。 1行スキップした後、3行を先頭に "# " を付与して出力しています。

実行結果

> java -cp .;sodium.jar ReadLineFile test1.txt

# 2b
# 3c
# 4d

test1.txt を処理した際の (a) ～ (d) の Event に対するイベント伝播状況をまとめると以下のようになっていると考えられます。（○ は伝播する、× は伝播しない）

対象行	値	(a)	(b)	(c)	(d)
1行目	1a	○	○	×	×
2行目	2b	○	○	○	○
3行目	3c	○	○	○	○
4行目	4d	○	○	○	○
5行目	5e	○	○	○	×

(d) まで到達した 2 ～ 4行目だけを出力する結果となります。

2014-11-23

Sodium で関数型リアクティブプログラミング

Java FRP

関数型リアクティブプログラミング（FRP）用のライブラリ Sodium を試してみました。

Sodium

Sodium には現時点で Java・Haskell・C++・C# 用のライブラリが用意されていますが（Embedded-C や Rust 用のライブラリも実装中の模様）、今回は Java 用のモジュールを使います。

今回のソースは http://github.com/fits/try_samples/tree/master/blog/20141123/

はじめに

Sodium の Java 用モジュールは Maven のセントラルリポジトリ等で配布されていないようなので、ソースを取得してビルドする事にします。

ビルドには Java 8 と Apache Ant を使います。（build.xml の source・target が 1.8 となっています）

ビルド例

$ git clone https://github.com/SodiumFRP/sodium.git
・・・
$ cd sodium/java
$ ant

ビルドに成功すると sodium/sodium.jar ファイルが生成されます。

Event と Behavior

Sodium では下記のような Event と Behavior を組み合わせて処理を組み立てます。

クラス	特徴	現在値の取得（sample メソッド）	イベント受信（listen メソッド）
Event	離散的なストリームを扱う	×	○
Behavior	連続的なストリームを扱う	○	×

Event の処理内容

まずは Event を単純に listen するだけの処理を実装してみます。

Event に何らかの値を送信する（イベントを発火させる）には Event のサブクラスである EventSink の send メソッドを使います。

なお、今回のようなサンプルでは Listener を unlisten する必要は無いのですが、一応入れています。

EventSample.java

import sodium.*;

class EventSample {
    public static void main(String... args) {

        EventSink<String> es = new EventSink<>();
        Listener esl = es.listen(System.out::println);

        es.send("ES1");

        System.out.println("---");

        es.send("ES2");

        esl.unlisten();
    }
}

実行結果は下記の通りです。

EventSink へ send した値（ES1 と ES2）が listen の処理（System.out::println）へ渡されています。

ビルドと実行

> javac -cp sodium.jar EventSample.java

> java -cp .;sodium.jar EventSample

ES1
---
ES2

Behavior の処理内容

次は、Behavior のカレント値の変更を listen してみます。

Behavior を直接 listen する事はできませんが、updates や value メソッドを使えば Behavior の値の変更に対応した Event を取得できます。

updates と value の違いは、取得した Event が listen 時にカレント値を含むかどうかの違いです。

動作としては RxJava の PublishSubject と BehaviorSubject にそれぞれ該当すると思います。

Behavior の Event 取得メソッド	listen 時のカレント値の扱い	RxJava の類似クラス
updates	含まない	rx.subjects.PublishSubject
value	含む	rx.subjects.BehaviorSubject

Event と同様に Behavior のカレント値を変更するには BehaviorSink の send メソッドを使います。

BehaviorSample.java

import sodium.*;

class BehaviorSample {
    public static void main(String... args) {
        updatesSample();

        System.out.println("");

        valueSample();
    }
    // updates メソッドのサンプル
    private static void updatesSample() {
        System.out.println("*** Behavior.updates sample ***");

        BehaviorSink<String> bh = new BehaviorSink<>("BH1");
        Listener bhl = bh.updates().listen( msg -> System.out.println("behavior: " + msg) );

        bh.send("BH2");

        System.out.println("---");

        bh.send("BH3");

        bhl.unlisten();
    }
    // value メソッドのサンプル
    private static void valueSample() {
        System.out.println("*** Behavior.value sample ***");

        BehaviorSink<String> bh = new BehaviorSink<>("BH1");
        Listener bhl = bh.value().listen( msg -> System.out.println("behavior: " + msg) );

        bh.send("BH2");

        System.out.println("---");

        bh.send("BH3");

        bhl.unlisten();
    }
}

value メソッドの場合のみ、初期値として設定した値（BH1）を出力しています。

実行結果

> java -cp .;sodium.jar BehaviorSample

*** Behavior.updates sample ***
behavior: BH2
---
behavior: BH3

*** Behavior.value sample ***
behavior: BH1
behavior: BH2
---
behavior: BH3

Event の各種メソッド

最後に Event クラスの map・merge・hold・snapshot メソッドを簡単に試してみます。

map

map メソッドによって元の Event で発火した値を加工した値を発火する Event を作成できます。

import sodium.*;

class EventMethodSample {
    public static void main(String... args) {
        mapSample();
        ・・・
    }

    private static void mapSample() {
        System.out.println("*** Event.map sample ***");

        EventSink<String> es = new EventSink<>();
        Listener esl = es.listen( msg -> System.out.println("event sink: " + msg) );

        // 元の値に !!! を付ける Event 作成
        Event<String> me = es.map( msg -> msg + "!!!" );
        Listener mel = me.listen( msg -> System.out.println("mapped event: " + msg) );

        es.send("ME1");
        es.send("ME2");

        mel.unlisten();
        esl.unlisten();
    }
    ・・・
}

ちなみに、上記では使っていませんが、Listener は append する事が可能です。（append で単一の Listener へまとめれば unlisten を個々に実施しなくても済みます）

実行結果

> java -cp .;sodium.jar EventMethodSample

*** Event.map sample ***
event sink: ME1
mapped event: ME1!!!
event sink: ME2
mapped event: ME2!!!
・・・

merge

merge メソッドによって二つの Event をマージできます。下記ではどちらの Event が発火しても発火する Event を作成しています。

import sodium.*;

class EventMethodSample {
    public static void main(String... args) {
        ・・・
        mergeSample();
        ・・・
    }
    ・・・
    private static void mergeSample() {
        System.out.println("*** Event.merge sample ***");

        EventSink<String> es1 = new EventSink<>();
        Listener es1l = es1.listen( msg -> System.out.println("event sink1: " + msg) );

        EventSink<String> es2 = new EventSink<>();
        Listener es2l = es2.listen( msg -> System.out.println("event sink2: " + msg) );

        Event<String> me = es1.merge(es2);
        Listener mel = me.listen( msg -> System.out.println("merged event: " + msg) );

        es1.send("ES1-1");

        System.out.println("---");

        es2.send("ES2-1");

        System.out.println("---");

        es1.send("ES1-2");

        mel.unlisten();
        es2l.unlisten();
        es1l.unlisten();
    }
    ・・・
}

実行結果

> java -cp .;sodium.jar EventMethodSample
・・・

*** Event.merge sample ***
event sink1: ES1-1
merged event: ES1-1
---
event sink2: ES2-1
merged event: ES2-1
---
event sink1: ES1-2
merged event: ES1-2

・・・

hold

hold メソッドによって Event の発火した値でカレント値が変化する Behavior を作成できます。

import sodium.*;

class EventMethodSample {
    public static void main(String... args) {
        ・・・
        holdSample();
        ・・・
    }
    ・・・
    private static void holdSample() {
        System.out.println("*** Event.hold sample ***");

        EventSink<String> es = new EventSink<>();
        Listener esl = es.listen( msg -> System.out.println("event sink: " + msg) );

        Behavior<String> bh = es.hold("BH1");
        Listener bhl = bh.value().listen( msg -> System.out.println("behavior: " + msg) );

        es.send("ES1");

        System.out.println("bh current value: " + bh.sample());

        System.out.println("---");

        es.send("ES2");

        System.out.println("bh current value: " + bh.sample());

        esl.unlisten();
        bhl.unlisten();
    }
    ・・・
}

bh の初期値は BH1 ですが、send した値（ES1 や ES2）によって sample メソッドの結果が変化しています。

実行結果

> java -cp .;sodium.jar EventMethodSample
・・・

*** Event.hold sample ***
behavior: BH1
event sink: ES1
behavior: ES1
bh current value: ES1
---
event sink: ES2
behavior: ES2
bh current value: ES2

・・・

snapshot

snapshot によって Event 発火時に任意の Behavior のカレント値を発火する Event を作成できます。

import sodium.*;

class EventMethodSample {
    public static void main(String... args) {
        ・・・
        snapshotSample();
    }
    ・・・
    private static void snapshotSample() {
        System.out.println("*** Event.snapshot sample ***");

        EventSink<String> es = new EventSink<>();
        Listener esl = es.listen( msg -> System.out.println("event sink: " + msg) );

        Behavior<Integer> bh = new Behavior<>(1);
        Listener bhl = bh.value().listen( msg -> System.out.println("behavior: " + msg) );

        Event<Integer> se = es.snapshot(bh);
        Listener sel = se.listen( i -> System.out.println("snapshot event: " + i) );

        es.send("ES1");

        System.out.println("bh current value: " + bh.sample());

        System.out.println("---");

        es.send("ES2");

        System.out.println("bh current value: " + bh.sample());

        sel.unlisten();
        esl.unlisten();
        bhl.unlisten();
    }
}

snapshot で作成した Event （se）は EventSink へ send した値（ES1 と ES2）に関わらず、bh のカレント値（1）を発火しています。

実行結果

> java -cp .;sodium.jar EventMethodSample
・・・

*** Event.snapshot sample ***
behavior: 1
event sink: ES1
snapshot event: 1
bh current value: 1
---
event sink: ES2
snapshot event: 1
bh current value: 1

・・・

snapshot した Event で発火するのは Behavior のカレント値であることを確認するため、上記の Behavior を BehaviorSink へ変更し ES2 を send する前にカレント値を 2 へ変更してみました。

import sodium.*;

class EventMethodSample {
    public static void main(String... args) {
        ・・・
        snapshotSample2();
    }
    ・・・
    private static void snapshotSample2() {
        ・・・
        // BehaviorSink へ変更
        BehaviorSink<Integer> bh = new BehaviorSink<>(1);
        Listener bhl = bh.value().listen( msg -> System.out.println("behavior: " + msg) );

        Event<Integer> se = es.snapshot(bh);
        ・・・
        System.out.println("---");

        // bh のカレント値を 2 へ変更
        bh.send(2);
        es.send("ES2");

        System.out.println("bh current value: " + bh.sample());
        ・・・
    }
}

Behavior のカレント値を 2 へ変更した後、snapshot の Event は 2 の値を発火している事を確認できます。

実行結果

> java -cp .;sodium.jar EventMethodSample
・・・

*** Event.snapshot sample2 ***
behavior: 1
event sink: ES1
snapshot event: 1
bh current value: 1
---
behavior: 2
event sink: ES2
snapshot event: 2
bh current value: 2

2014-11-12

Spark SQL で CSV ファイルを処理2 - GeoLite2

Scala Spark

前回の「Spark SQL で CSV ファイルを処理 - GeoLite Legacy」に続き、今回は Spark SQL を使って GeoLite2 City CSV ファイルを処理してみます。

GeoLite2

今回のソースは http://github.com/fits/try_samples/tree/master/blog/20141112/

はじめに

GeoLite2 City の CSV は下記のような 2種類のファイルで構成しています。

GeoLite2-City-Blocks.csv （IP と都市情報とのマッピング）
GeoLite2-City-Locations.csv （国・都市情報）

GeoLite2-City-Blocks.csv で IP アドレスから geoname_id を割り出し、GeoLite2-City-Locations.csv で geoname_id から国・都市を特定します。

ファイルの内容は下記のようになっており、IP は IPv6 の形式で記載されています。

GeoLite2-City-Blocks.csv の例

network_start_ip,network_prefix_length,geoname_id,registered_country_geoname_id,represented_country_geoname_id,postal_code,latitude,longitude,is_anonymous_proxy,is_satellite_provider
・・・
::ffff:1.0.64.0,114,1862415,1861060,,,・・・
・・・
2602:30a:2c1d::,48,5368361,,,・・・
・・・

GeoLite2-City-Locations.csv の例

geoname_id,continent_code,continent_name,country_iso_code,country_name,subdivision_iso_code,subdivision_name,city_name,metro_code,time_zone
1862415,AS,Asia,JP,Japan,34,Hiroshima,・・・
・・・

Spark SQL を使って IP アドレスから都市判定

GeoLite Legacy の Country CSV を処理した前回との違いは、下記 2点です。

(1) GeoLite2-City-Blocks.csv と GeoLite2-City-Locations.csv の 2つの CSV を geoname_id で join する
(2) network_start_ip と network_prefix_length を使って IP アドレスの数値の範囲を算出する

(1) は前回と同様に CSV を処理して SQL で join するだけです。 (2) は下記のようにして求める事ができます。

(a) IP アドレスの開始値は network_start_ip を数値化
(b) IP アドレスの終了値は (a) の値の下位 128 - network_prefix_length ビットを全て 1 とした値

今回は IPv4 のみを対象とするため、GeoLite2-City-Blocks.csv の ::ffff: で始まる行だけを使って（::ffff: 以降がそのまま IPv4 に該当）、上記 (a) と (b) の処理を実装してみました。

注意点として、GeoLite2-City-Locations.csv には subdivision_iso_code 以降が全て空欄のデータも含まれていました。（例えば 2077456,OC,Oceania,AU,Australia,,,,, を split(",") すると Array(2077456, OC, Oceania, AU, Australia) となってしまいます）

GetCity.scala

import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext

import java.net.InetAddress

// GeoLite2-City-Blocks.csv 用のスキーマ定義
case class IpMapping(startIpNum: Long, endIpNum: Long, geonameId: String)
// GeoLite2-City-Locations.csv 用のスキーマ定義
case class City(geonameId: String, country: String, city: String)

object GetCity extends App {
    if (args.length < 1) {
        println("<ip address>")
        System.exit(0)
    }

    // IPv4 の数値変換
    val toIpNum = (ip: String) => Integer.toUnsignedLong(InetAddress.getByName(ip).hashCode())

    val locationFile = "GeoLite2-City-Locations.csv"
    val blockFile = "GeoLite2-City-Blocks.csv"

    val sc = new SparkContext("local", "GetCity")

    val sqlContext = new SQLContext(sc)

    import sqlContext.createSchemaRDD

    val locations = sc.textFile(locationFile).map(_.split(",")).map { r =>
        // City 情報の無いデータ（subdivision_iso_code 以降が空欄）への対処
        val city = if (r.length > 7) r(7) else ""
        City(r(0), r(4), city)
    }

    locations.registerTempTable("locations")

    // IPv4 のみ （::ffff: で始まるもの） を対象
    val blocks = sc.textFile(blockFile).filter(_.startsWith("::ffff:")).map(_.split(",")).map { r =>
        val mask = -1 << (128 - r(1).toInt)
        // (a)
        val startIpNum = toIpNum(r(0).replaceAll("::ffff:", ""))
        // (b)
        val endIpNum = startIpNum | ~mask

        IpMapping(startIpNum, endIpNum, r(2))
    }

    blocks.registerTempTable("blocks")

    val ipNum = toIpNum(args(0))

    val rows = sqlContext.sql(s"""
        select
            city,
            country
        from
            locations lo
            join blocks bl on
                bl.geonameId = lo.geonameId
        where
            startIpNum <= ${ipNum} and
            endIpNum >= ${ipNum}
    """)

    rows.foreach( r => println(s"${r(0)}, ${r(1)}") )
}

上記では、IP の終了値 (b) を算出するために、上位ビットを 1、下位ビットを 0 にした mask を作成し、これをビット反転して開始値 (a) と論理和をとっています。

例えば、network_start_ip が ::ffff:1.0.64.0 で network_prefix_length が 114 のデータの場合、(a) の値は 1.0.64.0 を数値化して 16793600、mask 変数の値は 2進数で ・・・111100000000000000、(b) の値は mask 変数の値をビット反転した 011111111111111 と (a) の値との論理和で 16809983 となり、16793600 ～ 16809983 の範囲内にある IP アドレスが該当する事になります。

実行（Gradle 利用）

Gradle 2.1

前回と同様に Gradle で実行します。
slf4j-nop を使って Spark の標準的なログ出力を抑制している点も同じです。

build.gradle

apply plugin: 'application'
apply plugin: 'scala'

repositories {
    mavenCentral()
}

dependencies {
    compile 'org.scala-lang:scala-library:2.10.4'
    compile('org.apache.spark:spark-sql_2.10:1.1.0') {
        exclude module: 'slf4j-log4j12'
    }
    runtime 'org.slf4j:slf4j-nop:1.7.7'
}

mainClassName = 'GetCity'

run {
    if (project.hasProperty('args')) {
        args project.args.split(' ')
    }
}

実行結果1

> gradle run -q -Pargs=1.21.127.254

Tokyo, Japan

実行結果2

> gradle run -q -Pargs=223.255.254.1

, Singapore

2014-11-03

Spark SQL で CSV ファイルを処理 - GeoLite Legacy

Scala Spark

以前、H2 を使って CSV ファイルを SQL で処理しましたが、今回は Spark SQL を使ってみました。

Spark SQL

「IPアドレスから地域を特定する2 - GeoLite Legacy Country CSV」で使った GeoLite Legacy Country CSV を使って同様の処理を Spark SQL で実装します。

GeoLite Legacy

今回のソースは http://github.com/fits/try_samples/tree/master/blog/20141103-2/

Spark SQL を使って IP アドレスから国判定

Spark SQL で扱うテーブルのスキーマを定義する方法はいくつか用意されているようですが、今回はケースクラスをスキーマとして登録する方法で実装しました。

処理の手順は下記のようになります。

(1) スキーマ用のクラス定義
(2) CSV ファイルを処理して RDD 作成
(3) テーブル登録
(4) SQL の実行

(2) の処理で (1) のケースクラスを格納した RDD を作成し、(3) の処理で (2) で処理したオブジェクトをテーブルとして登録します。

(2) の処理までは通常の Spark の API を使った処理ですが、import sqlContext.createSchemaRDD によって (3) で registerTempTable メソッドを呼び出す際に RDD から Spark SQL の SchemaRDD へ暗黙変換が実施されます。

registerTempTable の引数としてテーブル名を渡す事で、SQL 内でこのテーブル名を使用できるようになります。

そのあとは SQL を実行して結果を出力するだけです。

foreach の要素となる org.apache.spark.sql.Row の実体は org.apache.spark.sql.catalyst.expressions.Row トレイトで、このトレイトが Seq トレイトを extends しているため head などの Seq の API も使えます。

GetCountry.scala

import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext

import java.net.InetAddress

// (1) スキーマ用のクラス定義
case class IpCountry(startIpNum: Long, endIpNum: Long, countryName: String)

object GetCountry extends App {
    if (args.length < 1) {
        println("<ip address>")
        System.exit(0)
    }

    val countryFile = "GeoIPCountryWhois.csv"

    val sc = new SparkContext("local", "GetCountry")

    val sqlContext = new SQLContext(sc)

    // RDD を SchemaRDD へ暗黙変換するための定義
    import sqlContext.createSchemaRDD

    // (2) CSV ファイルを処理して RDD 作成
    val countries = sc.textFile(countryFile).map(_.replaceAll("\"", "").split(",")).map { d =>
        IpCountry(d(2).toLong, d(3).toLong, d(5))
    }
    // (3) テーブル登録
    countries.registerTempTable("countries")

    val ipNum = Integer.toUnsignedLong( InetAddress.getByName(args(0)).hashCode )
    // (4) SQL 実行
    val rows = sqlContext.sql(s"""
        select
            countryName
        from
            countries
        where
            startIpNum <= ${ipNum} and
            endIpNum >= ${ipNum}
    """)

    rows.foreach( r => println(r.head) )
}

実行（Gradle 利用）

Gradle 2.1

今回は Gradle で実行するため、下記のようなビルド定義ファイルを用意しました。

現時点では、Maven のセントラルリポジトリに Scala 2.11 用の Spark SQL の JAR ファイルは用意されていないようなので、Scala 2.10.4 を使います。

今回の用途では Spark の標準的なログ出力が邪魔だったので slf4j-log4j12 の代わりに slf4j-nop を使うようにしてログ出力を抑制しました。

build.gradle

apply plugin: 'application'
apply plugin: 'scala'

repositories {
    mavenCentral()
}

dependencies {
    compile 'org.scala-lang:scala-library:2.10.4'
    compile('org.apache.spark:spark-sql_2.10:1.1.0') {
        // Spark のログ出力を抑制
        exclude module: 'slf4j-log4j12'
    }
    runtime 'org.slf4j:slf4j-nop:1.7.7'
}

mainClassName = 'GetCountry'

run {
    if (project.hasProperty('args')) {
        // コマンドライン引数の設定
        args project.args.split(' ')
    }
}

更に、Gradle のログ出力（タスクの実行経過）も抑制したいので、-q オプションを使って実行しました。

実行結果1

> gradle run -q -Pargs=1.21.127.254

Japan

実行結果2

> gradle run -q -Pargs=223.255.254.1

Singapore

2014-11-03

Java アプリケーションの起動スクリプト - Windows・Linux

Java

指定ディレクトリ内の全 JAR ファイルをクラスパスへ指定して Java アプリケーションを起動するスクリプト例です。

今回のソースは http://github.com/fits/try_samples/tree/master/blog/20141103-1/

Windows の場合（bat）

lib ディレクトリ内の全 JAR ファイルをクラスパスへ指定して Java アプリケーション（下記では sample.SampleApp）を実行する bat スクリプトは下記のようになります。

enabledelayedexpansion と !<環境変数>! で環境変数を遅延展開させる点に注意が必要です。

こうしないと CP 環境変数へ JAR のパスが正しく設定されません。

exec_java_sample.bat

@echo off

setlocal enabledelayedexpansion

set APP_CLASS=sample.SampleApp

set BASE_DIR=%~d0%~p0

set CP=%BASE_DIR%
set LIB=%BASE_DIR%lib

for %%j in ("%LIB%\*.jar") do call set CP=!CP!;%%j

java -cp %CP% %APP_CLASS% %*

endlocal

Linux の場合（bash）

bash の場合も bat とほぼ同じ様に書けます。

exec_java_sample.sh

#!/bin/sh

APP_CLASS=sample.SampleApp

BASE_DIR=`dirname $0`

CP=$BASE_DIR
LIB=$BASE_DIR/lib

for jar in $LIB/*.jar; do CP=$CP:$jar; done

java -cp $CP $APP_CLASS $*

Web アプリケーションの作成

ファイル構成

ビルド定義

build.gradle

(1) HandlerFactory 実装クラスの作成

src/main/java/AppHandlerFactory.java

(2) handlerFactory の設定

src/ratpack/ratpack.properties

テスト実行

成果物の生成

MyBatis の場合

mybatis_sql_gen.groovy

実行

mapper.xml

実行例1

実行例2

iBatis の場合

ibatis_sql_gen.groovy

実行

mapper.xml

実行例1

実行例2

skip・take 処理

ReadLineFile.java

unlisten する処理を追加した batch メソッドの例

実行

test1.txt

実行結果

はじめに

ビルド例

Event と Behavior

Event の処理内容

EventSample.java

ビルドと実行

Behavior の処理内容

BehaviorSample.java

実行結果

Event の各種メソッド

map

実行結果

merge

実行結果

hold

実行結果

snapshot

実行結果

実行結果

はじめに

GeoLite2-City-Blocks.csv の例

GeoLite2-City-Locations.csv の例

Spark SQL を使って IP アドレスから都市判定

GetCity.scala

実行 （Gradle 利用）

build.gradle

実行結果1

実行結果2

Spark SQL を使って IP アドレスから国判定

GetCountry.scala

実行 （Gradle 利用）

build.gradle

実行結果1

実行結果2

Windows の場合 （bat）

exec_java_sample.bat

Linux の場合 （bash）

exec_java_sample.sh

実行（Gradle 利用）

実行（Gradle 利用）

Windows の場合（bat）

Linux の場合（bash）