Запрос к сжатому CSV-файлу на AWS S3

Размер файла может стать большим на S3. Я писал API для получения данных на S3, которые соответствуют условию в SQL-запросе, и возникает ошибка, говорящая, что размер файла слишком велик для использования selectObjectContent из S3 SDK.

Один из способов обработки — запросить его сжатую версию.

Во-первых, давайте использовать GZIP для сжатия файла.

with gzip.GzipFile('data.gz', 'w') as cf:
    cf.write(bytes(json.dumps(payload, default=str).encode('utf-8')))
# data is a list of json [{key: val]}, {key, val}, ...]

Затем загрузите файл на S3.

s3.Bucket('bucket-name').upload_file('data.gz',
                                    'file-path-on-s3/data.gz')

Теперь давайте запустим SQL-запрос к этому GZIP-файлу на S3, используя SDK, предоставленный AWS S3. Я использовал SDK в javascript.

Вы можете обнаружить, что поток событий прерывается в неправильном месте, поэтому выходные данные трудно анализировать. Я использую функцию-обработчик для объединения возвращаемого списка строк, что-то вроде нескольких сломанных фрагментов строки json, а затем разделяю их по символу «\n», который является разделителем записей в полезной нагрузке.

Вывод будет выглядеть как обычный объект json.

const s3 = new AWS.S3({
  region: 'your region',
  accessKeyId: 'your key id',
  secretAccessKey: 'your secret key',
  httpOptions: {timeout: 2000, connectTimeout: 2000},
});
s3.selectObjectContent({
    Bucket: 'bucket-name',
    Key: 'file-path-on-s3/data.gz',
    Expression: `SELECT * FROM S3Object[*][*] s WHERE [Your SQL Condition]`,
    ExpressionType: 'SQL',
    InputSerialization: {'JSON': {'Type': 'DOCUMENT'}, 'CompressionType': 'GZIP',},
    OutputSerialization: {'JSON': {}},
  }
).promise()
  .then(data => jsonHandler(data, res))
  .catch(e => res.status(500).send({err: e.toString()}))
const jsonHandler = (output, res) => new Promise((resolve, reject) => {
  const eventStream = output.Payload;
  let jsonData = null;
  let stream = '';
  eventStream.on('data', event => {
    if (event.Records) {
      // records expected: {}\n OR {}\n...{}\n
      const records = get(event, 'Records.Payload', '').toString();
      //join two partial payloads(string) which are split at incorrect position and split them in correct position after stream ends
      stream = stream.concat(records);
    } else if (event.End) {
      let stringArr = stream.split('\n')
      //the last el is ""
      var filtered = stringArr.filter(function (el) {
        return el !== '';
      });

      jsonData = filtered.map(d => JSON.parse(d));

      return resolve(res.status(200).send({data: jsonData}));
    }
  });