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

MyBatis / iBatisAPI を使って 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       (         ?       ,     ?       ,         ?       )

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

前回に続き、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行目だけを出力する結果となります。

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

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

Sodium には現時点で JavaHaskellC++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 では下記のような EventBehavior を組み合わせて処理を組み立てます。

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

Event の処理内容

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

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

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

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 した値 (ES1ES2) が listen の処理 (System.out::println) へ渡されています。

ビルドと実行
> javac -cp sodium.jar EventSample.java

> java -cp .;sodium.jar EventSample

ES1
---
ES2

Behavior の処理内容

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

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

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

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

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

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

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();
    }
    ・・・
}

ちなみに、上記では使っていませんが、Listenerappend する事が可能です。 (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 した値 (ES1ES2) によって 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 した値 (ES1ES2) に関わらず、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

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

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

今回のソースは 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

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

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

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

今回のソースは 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 SQLSchemaRDD へ暗黙変換が実施されます。

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

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

指定ディレクトリ内の全 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 $_`

CP=$BASE_DIR
LIB=$BASE_DIR/lib

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

java -cp $CP $APP_CLASS $*

Vagrant で VirtualBox 仮想マシンの CPU 数とメモリサイズを変更

VagrantVirtualBox 上の仮想マシンの CPU 数とメモリサイズを変更する方法としては、現時点で下記 2通りがあります。

  • (a) cpus や memory を使用する
  • (b) customize を使用する

ちなみに、Vagrant ソースの plugins/providers/virtualbox/config.rb を見ると、(a) は内部で customize を呼び出して (b) と同じ事をしているだけです。

(a) cpus や memory を使用する

cpusmemory を使って CPU 数とメモリサイズを変更する Vagrantfile は下記のようになります。

  • CPU 数を 2
  • メモリを 1 GB (1024 MB)
Vagrantfile 設定例 (a)
・・・
Vagrant.configure(VAGRANTFILE_API_VERSION) do |config|
    ・・・
    config.vm.provider "virtualbox" do |v|
        v.cpus = 2
        v.memory = 1024
    end
end

基本的に、こちらの方が分かり易いと思います。

(b) customize を使用する

customize を使って CPU 数とメモリサイズを変更する Vagrantfile は下記のようになります。

  • CPU 数を 2
  • メモリを 1 GB (1024 MB)
Vagrantfile 設定例 (b)
・・・
Vagrant.configure(VAGRANTFILE_API_VERSION) do |config|
    ・・・
    config.vm.provider "virtualbox" do |v|
        v.customize ["modifyvm", :id, "--cpus", "2"]
        v.customize ["modifyvm", :id, "--memory", "1024"]
    end
end

なお、上記の customize は、:id の箇所を変換した引数を VBoxManage コマンドへ渡して実行する事になります。 (Vagrant ソースの plugins/providers/virtualbox/action/customize.rb 等を参照)

つまり、仮想マシン 5f6002fc-1b3c-1523-5be3-c143831fbe05 に対して下記のようなコマンドを実行するのと同じです。

VBoxManage コマンドを使って CPU 数とメモリサイズ変更する例
> VBoxManage modifyvm 5f6002fc-1b3c-1523-5be3-c143831fbe05 --cpus 2
> VBoxManage modifyvm 5f6002fc-1b3c-1523-5be3-c143831fbe05 --memory 1024