概要

Elasticsearchのバルクロードは次の形式なのですが、ふと必要にかられてこんな形の2行1ペアのJSON Linesファイルから、なんちゃってselectを行うトイプログラムを作成してみました。

PUT hockey/_bulk?refresh
{"index":{"_id":1}}
{"first":"johnny","last":"gaudreau","goals":[9,27,1],"assists":[17,46,0],"gp":[26,82,1],"born":"1993/08/13"}
レコード分繰り返し

↓バルクロードの形式の例の記載あり（以降の記事はリンク先のバルクロード例のJSONLinesファイルを「esjson.jsonlines」ファイルに保存している体で記載しています） www.elastic.co

実例(Python)

import sys
import json
import re

def getobj(dict,keystr): 
    wrk = dict
    for i in keystr.split('.'):
            wrk = wrk[i]
    return wrk

def my_print_2json(i,j,stderr=False):
    i_json = json.dumps(i,ensure_ascii=False)
    j_json = json.dumps(j,ensure_ascii=False)
    if stderr:
        print('STDERR\t' + i_json,file=sys.stderr)
        print('STDERR\t' + j_json,file=sys.stderr)
    else:
        print(i_json)
        print(j_json)

def my_print_json(i):
    i_json = json.dumps(i,ensure_ascii=False)
    print(i_json)

if __name__ == "__main__":
    
    mode = sys.argv[1] #both/either/each
    
    ii = sys.argv[2]
    i_re = re.compile(sys.argv[3])
    
    jj = sys.argv[4]
    j_re = re.compile(sys.argv[5])
    
    jsons = []
    for l in sys.stdin:
        jsons.append(json.loads(l))
    

    for i, j in zip(jsons[0::2], jsons[1::2]):
        cond1 = re.findall(i_re, str(getobj(i,ii)))
        cond2 = re.findall(j_re, str(getobj(j,jj)))
        if mode == 'both' and (cond1 and cond2):
            my_print_2json(i,j)
        elif mode == 'either' and (cond1 or cond2):
            my_print_2json(i,j)
        elif mode == 'each':
            if cond1: my_print_json(i)
            if cond2: my_print_json(j)
        else:
            my_print_2json(i,j,stderr=True)

こんな感じで起動〜出力が得られます。

foo$ cat esjson.jsonlines | python3 esjsonline_select.py both index._id '^1'  first '^jo'
{"index": {"_id": 1}}
{"first": "johnny", "last": "日本語gaudreau", "goals": [9, 27, 1], "assists": [17, 46, 0], "gp": [26, 82, 1], "born": "1993/08/13"}
STDERR  {"index": {"_id": 2}}
STDERR  {"first": "sean", "last": "monohan", "goals": [7, 54, 26], "assists": [11, 26, 13], "gp": [26, 82, 82], "born": "1994/10/12"}
STDERR  {"index": {"_id": 3}}
STDERR  {"first": "jiri", "last": "hudler", "goals": [5, 34, 36], "assists": [11, 62, 42], "gp": [24, 80, 79], "born": "1984/01/04"}
STDERR  {"index": {"_id": 4}}
STDERR  {"first": "micheal", "last": "frolik", "goals": [4, 6, 15], "assists": [8, 23, 15], "gp": [26, 82, 82], "born": "1988/02/17"}
STDERR  {"index": {"_id": 5}}
STDERR  {"first": "sam", "last": "bennett", "goals": [5, 0, 0], "assists": [8, 1, 0], "gp": [26, 1, 0], "born": "1996/06/20"}
STDERR  {"index": {"_id": 6}}
STDERR  {"first": "dennis", "last": "wideman", "goals": [0, 26, 15], "assists": [11, 30, 24], "gp": [26, 81, 82], "born": "1983/03/20"}
STDERR  {"index": {"_id": 7}}
STDERR  {"first": "david", "last": "jones", "goals": [7, 19, 5], "assists": [3, 17, 4], "gp": [26, 45, 34], "born": "1984/08/10"}
STDERR  {"index": {"_id": 8}}
STDERR  {"first": "tj", "last": "brodie", "goals": [2, 14, 7], "assists": [8, 42, 30], "gp": [26, 82, 82], "born": "1990/06/07"}
STDERR  {"index": {"_id": 39}}
STDERR  {"first": "mark", "last": "giordano", "goals": [6, 30, 15], "assists": [3, 30, 24], "gp": [26, 60, 63], "born": "1983/10/03"}
STDERR  {"index": {"_id": 10}}
STDERR  {"first": "mikael", "last": "backlund", "goals": [3, 15, 13], "assists": [6, 24, 18], "gp": [26, 82, 82], "born": "1989/03/17"}
{"index": {"_id": 11}}
{"first": "joe", "last": "colborne", "goals": [3, 18, 13], "assists": [6, 20, 24], "gp": [26, 67, 82], "born": "1990/01/30"}

第1引数 → both/either/each 第2引数と第3引数、　および　第4引数と第5引数 → それぞれ前者でJSONのフィールド名を「A.B.C」形式で指定して、その値がマッチすべきPythonのreモジュールの正規表現を指定します。　前の組が奇数行、後ろの組がその次行の偶数行のJSON に対するselect演算相当になります。

ここで、bothはカップリングの行の両方が条件に該当した場合にカップルの2つの行を出力、eitherはどちらかの行がマッチした場合にそのカップルの2つの行を出力、eachは単に1行ずつのマッチングとなっています。

おわりに

AWS Athenaやjqコマンドなども使えない淋しい環境においては、意外に使える（応用がきく）かもしれず、思わず記事にしてしまいました。

2行一組で、diffというかcompare、sortなども便利かも...そのうちメモっておこう（独り言）

↓ 続きです。

itdepends.hateblo.jp

はてだBlog（仮称）

Elasticsearch のバルクロード用JSON Lines ファイルをselectする toy スクリプト

概要

実例(Python)

おわりに