AWS Athena экспортирует массив структур в JSON

У меня есть таблица Athena, где некоторые поля имеют довольно сложный вложенный формат. Записями поддержки в S3 являются JSON. Вдоль этих строк (но у нас есть еще несколько уровней гнездования):

CREATE EXTERNAL TABLE IF NOT EXISTS test (
  timestamp double,
  stats array<struct<time:double, mean:double, var:double>>,
  dets array<struct<coords: array<double>, header:struct<frame:int, 
    seq:int, name:string>>>,
  pos struct<x:double, y:double, theta:double>
)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
WITH SERDEPROPERTIES ('ignore.malformed.json'='true')
LOCATION 's3://test-bucket/test-folder/'

Теперь нам нужно иметь возможность запрашивать данные и импортировать результаты в Python для анализа. Из-за ограничений безопасности я не могу напрямую подключиться к Athena; Мне нужно дать кому-то запрос, и тогда они дадут мне результаты CSV.

Если мы просто сделаем прямой выбор *, мы вернем столбцы struct/array в формате, который не совсем JSON. Здесь пример ввода входного файла:

{"timestamp":1520640777.666096,"stats":[{"time":15,"mean":45.23,"var":0.31},{"time":19,"mean":17.315,"var":2.612}],"dets":[{"coords":[2.4,1.7,0.3], "header":{"frame":1,"seq":1,"name":"hello"}}],"pos": {"x":5,"y":1.4,"theta":0.04}}

И пример вывода:

select * from test

"timestamp","stats","dets","pos"
"1.520640777666096E9","[{time=15.0, mean=45.23, var=0.31}, {time=19.0, mean=17.315, var=2.612}]","[{coords=[2.4, 1.7, 0.3], header={frame=1, seq=1, name=hello}}]","{x=5.0, y=1.4, theta=0.04}"

Я надеялся, что эти вложенные поля экспортируются в более удобный формат - получение их в JSON было бы здорово.

К сожалению, похоже, что приведение в JSON работает только для карт, а не структур, потому что оно просто сглаживает все в массивы:

SELECT timestamp, cast(stats as JSON) as stats, cast(dets as JSON) as dets, cast(pos as JSON) as pos FROM "sampledb"."test"

"timestamp","stats","dets","pos"
"1.520640777666096E9","[[15.0,45.23,0.31],[19.0,17.315,2.612]]","[[[2.4,1.7,0.3],[1,1,""hello""]]]","[5.0,1.4,0.04]"

Есть ли хороший способ конвертировать в JSON (или другой удобный для импорта формат), или я должен просто идти вперед и выполнять пользовательскую функцию синтаксического анализа?

Ответ 1

Я просмотрел всю документацию и, к сожалению, сейчас нет способа сделать это. Единственным возможным обходным решением является

преобразование структуры в json при запросе athena

SELECT
  my_field,
  my_field.a,
  my_field.b,
  my_field.c.d,
  my_field.c.e
FROM 
  my_table

Или я бы преобразовал данные в json, используя пост-обработку. Ниже сценария показано, как

#!/usr/bin/env python
import io
import re

pattern1 = re.compile(r'(?<={)([a-z]+)=', re.I)
pattern2 = re.compile(r':([a-z][^,{}. [\]]+)', re.I)
pattern3 = re.compile(r'\\"', re.I)

with io.open("test.csv") as f:
    headers = list(map(lambda f: f.strip(), f.readline().split(",")))
    for line in f.readlines():
        orig_line = line
        data = []
        for i, l in enumerate(line.split('","')):
            data.append(headers[i] + ":" + re.sub('^"|"$', "", l))

        line = "{" + ','.join(data) + "}"
        line = pattern1.sub(r'"\1":', line)
        line = pattern2.sub(r':"\1"', line)
        print(line)

Выходные данные на ваших входных данных

{"timestamp":1.520640777666096E9,"stats":[{"time":15.0, "mean":45.23, "var":0.31}, {"time":19.0, "mean":17.315, "var":2.612}],"dets":[{"coords":[2.4, 1.7, 0.3], "header":{"frame":1, "seq":1, "name":"hello"}}],"pos":{"x":5.0, "y":1.4, "theta":0.04}
}

Что является действительным JSON

Ответ 2

Код python от @tarun почти достал меня, но мне пришлось изменить его несколькими способами из-за моих данных. В частности, у меня есть:

json-структуры, сохраненные в Афине как строки
Строки, которые содержат несколько слов, и поэтому должны находиться между двойными кавычками. Некоторые из них содержат символы "[]" и "{}".

Вот код, который работал для меня, надеюсь, будет полезен для других:

#!/usr/bin/env python
import io
import re, sys

pattern1 = re.compile(r'(?<={)([a-z]+)=', re.I)
pattern2 = re.compile(r':([a-z][^,{}. [\]]+)', re.I)
pattern3 = re.compile(r'\\"', re.I)

with io.open(sys.argv[1]) as f:
    headers = list(map(lambda f: f.strip(), f.readline().split(",")))
    print(headers)
    for line in f.readlines():

        orig_line = line
        #save the double quote cases, which mean there is a string with quotes inside
        line = re.sub('""', "#", orig_line)
        data = []
        for i, l in enumerate(line.split('","')):
            item = re.sub('^"|"$', "", l.rstrip())
            if (item[0] == "{" and item[-1] == "}") or (item[0] == "[" and item[-1] == "]"):
                data.append(headers[i] + ":" + item)
            else: #we have a string
                data.append(headers[i] + ": \"" + item + "\"")

        line = "{" + ','.join(data) + "}"
        line = pattern1.sub(r'"\1":', line)
        line = pattern2.sub(r':"\1"', line)

        #restate the double quotes to single ones, once inside the json
        line = re.sub("#", '"', line)

        print(line)