csvを読んでelasticsearchのバルクインサート用のjsonに整形する

elasticsearchにバルクインサートする時、
データのjson1行につきコマンドのjsonを1行追加しないとならないので。

stream2es だと一気に登録までしてしまうので、
変換の時点で一度確認したかったので書いてみた。

中身は

1行目にヘッダ。
それ以降がデータ。

エスケープなどはRFC4180 っぽい感じ。

items.txt

"item_id","title","description","price","url","image_url"
"11111","たいとる","でぃすくりぷしょん","1000","http://example.com/item/11111","http://example.com/11111.jpg"
"22222","たいとる2","でぃすくりぷしょん2","2000","http://example.com/item/22222","http://example.com/22222.jpg"

変換

rubyのCSVクラスがちゃんとcsv読んでくれるので、これを使うことにした。

が、パースにけっこう時間がかかるみたいなので、
大量の件数(百万件単位とか)扱う時は速度面で不安がある。

cmd2json.rb

#!/usr/bin/env ruby

require 'csv'
require 'json'
require 'securerandom'

line = STDIN.gets.chomp
csv = CSV.new(line)
header = csv.to_a[0]

INDEX = "test"
TYPE  = "type1"

CSV(STDIN).each_with_index do |row, i|
  index = { "index" =>
    { "_index" => INDEX, "_type" => TYPE, "_id" => SecureRandom.uuid }
  }
  puts JSON.dump(index)

  hash = Hash[header.zip row]
  puts JSON.dump(hash)
end

bash

cat items.txt | ./csv2json.rb > items.es.json

esに登録

で、こう。

bulk.sh

curl -s -XPOST localhost:9200/_bulk --data-binary @items.es.json

プロトタイプ

最初

csv => json
json => elasticsearch登録用のjson

ってやってみたけど、別に一気にやっちゃってもいいじゃん、
ってことで上記の形に落ち着いた。

2. の部分はbashでも書いたので、一応それも。
要 jq 。

1行ごとに jq 2回叩くので、
件数が多くなってくるとプロセスの起動と破棄コストがけっこう馬鹿にならなくなってくるため、あまりオススメはしない。
10万件くらい流し込んだだけでもけっこう時間かかった。

cmd2es.sh

#!/bin/bash

set -u

_INDEX=$1
_TYPE=$2

while read -r line; do
    _ID=`uuidgen`
    cat <<EOD | jq -c .
{
    "index" : {
        "_index" : "$_INDEX",
        "_type" : "$_TYPE",
        "_id" : "$_ID"
    }
}
EOD
    echo $line | jq . -c
done

bash

cat items.json | ./cmd2es.sh foo bar > items.es.json