Treasure Dataで活用できる技術のまとめ #hadoop

time項目の活用でlimit offsetを実現する

treasure dataのテーブルを作成する時、time項目がデフォルトが入っています。

やりたいこと：1億レコードを１００万件単位で分けて取得する

取得件数を指定して、巨大なデータを分けて取得の場合、sqlやmysqlはlimit offsetを使えば簡単に実現できますが、Treasure Dataのhiveやpresto「LIMIT ALL and OFFSET」がサポートされないです。

You can do by ROW_NUMBER in hive

SELECT *,ROW_NUMBER over (Order by id)  as rowid FROM mytable
where rowid > 0 and rowid <=20;
next time you have to change the condition in where clause.

SELECT *,ROW_NUMBER over (Order by id)  as rowid FROM mytable
    where rowid > 20 and rowid <=40;

hiveでこんなやり方ができるみたいですが、treasure dataの場合巨大なデータを条件的に取得する時、指定条件はtime項目以外の場合、速度がすごく重いです。上記のやり方で、3000万のレコードを試したところ、jobが裏側に何回リトライされましたが、永遠に終わらないです。

実現方法

timeのtypeはintなので、timeを数字で作成してテーブルに追加すれば、idとして使えます。
例：テーブルstaffsに1億レコードがあります。
Schema：time(int) id(int) name(string)

|time|id|name|
|--|--|--|
|1489639376|1|jane|
|1489639376|2|jone|
|1489639376|3|kitty|
|1489639376|4|may|
...

staffsのレコード数を取得

totalCount = select count(1) from staffs;

100万件単位で分割数計算 divisionNum = ceil(totalCount / 1000000)
tmp_staffsテーブル作成、Schemaはstaffsと同じにする
timeを数字で作成して、staffsのレコードを取得してtmp_staffsに追加

//divisionNumは100の場合
INSERT
  INTO
    tmp_staffs SELECT 
      CAST(
        FLOOR(1 + RAND()* 100) AS BIGINT
      ) AS time,
      id,
      name
    FROM
      staffs
;

tmp_staffsのレコードは下記になります。

|time|id|name|
|--|--|--|
|1|1|jane|
|1|2|jone|
|2|3|kitty|
|3|4|may|
...

※　time = 1 は'1970-01-01 09:00:01  JST'となります。
　 time = 2 は'1970-01-01 09:00:02  JST'となります。

tmp_staffsのtimeを範囲(1秒ずつ)で指定して取得する、結果は100万件ぐらい

SELECT 
  id,
  name
FROM
  tmp_staffs
WHERE
  TD_TIME_RANGE(time,
    '1970-01-01 09:00:01 JST',
    TD_TIME_ADD('1970-01-01 09:00:01',
      '1s',
      'JST'))

Array columnの活用でレコード数削減

Treasure Dataのテーブルのデータタイプはarrayのカラムがあります。レコードをまとめて保存できるし、hiveやprestoなど配列のfunctions、stringを配列に変換するfucntionsがいっぱいあります。Hive Built-in Functions、 Array Functions in Presto、 LanguageManual UDF

例：テーブルstaffs
Schema：time(int) id(int) name(string) access_date(string)

レコード①

|time|id|name|access_date|
|--|--|--|
|1489639376|1|jane|2017-03-01|
|1489639376|2|jone|2017-03-01|
|1489639376|3|kitty|2017-03-01|
|1489639376|4|may|2017-03-01|
|1489634376|1|jane|2017-03-05|
|1489634376|1|jane|2017-03-15|
|1489634376|2| jone |2017-03-10|

access_dateのデータタイプはarrayに変換にしたら、

レコードは下記になります。

レコード②

|time|id|name|access_date|
|--|--|--|
|1489639376|1|jane|["2017-03-01", "2017-03-05", "2017-03-15"]|
|1489639376|2|jone|["2017-03-01", "2017-03-10"]|
|1489639376|3|kitty|["2017-03-01"]|
|1489639376|4|may|["2017-03-01"]|

collect_setでstringからarrayに変更(hive)

SELECT 
  id,
  name,
  collect_set(access_date) AS access_date
FROM
  staffs
GROUP BY
  id,
  name

結果はレコード①からレコード②に変更。

concat_wsでarrayからstringに変更(hive)

SELECT 
  id,
  name,
  concat_ws(',', access_date) AS access_date
FROM
  staffs
GROUP BY
  id,
  name

レコード②から下記に変換する

|time|id|name|access_date|
|--|--|--|
|1489639376|1|jane|2017-03-01, 2017-03-05, 2017-03-15|
|1489639376|2|jone|2017-03-01, 2017-03-10|
|1489639376|3|kitty|2017-03-01|
|1489639376|4|may|2017-03-01|

array_max、array_min（presto）配列のMAX、MINの取得

UNNESTで二重キーにarrayカラムから最大値や最小値の取得

例：

|time|id|name|access_date|
|--|--|--|
|1489639376|1|jane|["2017-03-01", "2017-03-05", "2017-03-15"]|
|1485639376|1|jane|["2017-03-16"]|
|1485639376|1|jane|["2017-03-17"]|

最小値取得

SELECT 
  t1.id,
  t1.name,
  MIN(t1.access_date) AS access_date
FROM (
    SELECT 
      id,
      name,
      access_date
    FROM
      staffs CROSS
    JOIN
      UNNEST(access_date) AS t(access_date)
  ) t1
GROUP BY
 t1.id,
 t1.name

結果：

|time|id|name|access_date|
|--|--|--|
|1489639376|1|jane|2017-03-01|

最大値取得

SELECT 
  t1.id,
  t1.name,
  MAX(t1.access_date) AS access_date
FROM (
    SELECT 
      id,
      name,
      access_date
    FROM
      staffs CROSS
    JOIN
      UNNEST(access_date) AS t(access_date)
  ) t1
GROUP BY
 t1.id,
 t1.name

結果：

|time|id|name|access_date|
|--|--|--|
|1489639376|1|jane|2017-03-17|

条件により最小値取得

SELECT 
  t1.id,
  t1.name,
  MIN(t1.access_date) AS access_date
FROM (
    SELECT 
      id,
      name,
      access_date
    FROM
      staffs CROSS
    JOIN
      UNNEST(access_date) AS t(access_date)
  ) t1
WHERE 
 t1.access_date > '2017-03-10'
GROUP BY
 t1.id,
 t1.name

結果：

|time|id|name|access_date|
|--|--|--|
|1489639376|1|jane|2017-03-15|

タイムゾーンのフォーマットをJSTに指定すること

Treasure DataのHive UDFs ,Presto Native UDFsのdefault_timezone = 'UTC'になってますが、テーブルのレコードを保存する時、timeの項目がデフォルトフォーマットはJSTで保存されています。だからTD_TIME_ADD、TD_TIME_PARSEの関数を使う時、'JST'を指定したほうがいいです。

|日付|timeフォーマットUTC|timeフォーマットJST|
|--|--|--|
|11/27日|1480204800|1480172400|
|11/26日|1480118400|1480086000|
|11/25日|1480032000|1479999600|
|11/24日|1479945600|1479913200|
|11/23日|1479859200|1479826800|