:Processar dados do OSS armazenados em formatos de código aberto

Para dados armazenados em formatos de código aberto, o framework unstructured chama o método de implementação fornecido pela comunidade open source para analisar as informações. Esse método se integra perfeitamente ao MaxCompute para ler e interpretar dados nesses formatos.

Sintaxe para criar uma tabela externa

DROP TABLE [IF EXISTS] <external_table>;
CREATE EXTERNAL TABLE [IF NOT EXISTS] <external_table>
(<column schemas>)
[PARTITIONED BY (partition column schemas)]
[ROW FORMAT SERDE '<serde class>'
  [WITH SERDEPROPERTIES ('odps.properties.rolearn'='${roleran}' [,'name2'='value2',...])]
]
STORED AS <file format>
LOCATION 'oss://${endpoint}/${bucket}/${userfilePath}/';

A sintaxe é semelhante à do Hive. Observe os seguintes pontos:

• column schemas: defina as colunas da tabela externa. A definição das colunas deve ser idêntica à definição dos dados armazenados no OSS.

• ROW FORMAT SERDE: esta cláusula é necessária apenas quando você usa formatos especiais, como TEXTFILE.

• WITH SERDEPROPERTIES: se houver autorização STS no OSS, esta cláusula será obrigatória para definir a propriedade odps.properties.rolearn. Essa propriedade corresponde ao Alibaba Cloud Resource Name (ARN) da função do RAM. Configure STORED AS <file format> e <serde class> para especificar o formato de arquivo. O exemplo a seguir usa o formato ORC:

  CREATE EXTERNAL TABLE [IF NOT EXISTS] <external_table>
  (<column schemas>)
  [PARTITIONED BY (partition column schemas)]
  ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.orc.OrcSerde'
  WITH SERDEPROPERTIES ('odps.properties.rolearn'='${roleran}'
  STORED AS ORC
  LOCATION 'oss://${endpoint}/${bucket}/${userfilePath}/'

  Nota

  Se não houver autorização STS, não será necessário definir a propriedade

  odps.properties.rolearn

  . Basta especificar o

  AccessKey ID

  e o

  AccessKey secret

  na cláusula Location em texto simples.

• STORED AS: exclusivo para leitura de dados armazenados em formatos de código aberto. Difere da cláusula STORED BY usada para criar uma tabela externa unstructured padrão.

  STORED AS é seguido pelo formato de arquivo, como ORC, PARQUET, RCFILE, SEQUENCEFILE ou TEXTFILE. Na cláusula STORED AS, o tamanho de um arquivo não pode exceder 3 GB. Caso o arquivo ultrapasse esse limite, divida-o.

  Mapeamentos entre

  formatos de arquivo

  e

  classes SerDe

  :

  • SEQUENCEFILE: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe

  • TEXTFILE: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe

  • RCFILE: org.apache.hadoop.hive.serde2.columnar.LazyBinaryColumnarSerDe

  • ORC: org.apache.hadoop.hive.ql.io.orc.OrcSerde

  • ORCFILE: org.apache.hadoop.hive.ql.io.orc.OrcSerde

  • PARQUET: org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe

  • AVRO: org.apache.hadoop.hive.serde2.avro.AvroSerDe

• LOCATION: se o serviço OSS estiver associado, configure o AccessKey ID e o AccessKey secret em texto simples. Exemplo:

  LOCATION 'oss://${accessKeyId}:${accessKeySecret}@${endpoint}/${bucket}/${userPath}/'

  Não é possível usar tabelas externas para acessar dados do OSS por meio do endpoint público do serviço.

Criar uma tabela externa com base em um objeto PARQUET em um bucket do OSS

CREATE EXTERNAL TABLE tpch_lineitem_parquet
(
  l_orderkey bigint,
  l_partkey bigint,
  l_suppkey bigint,
  l_linenumber bigint,
  l_quantity double,
  l_extendedprice double,
  l_discount double,
  l_tax double,
  l_returnflag string,
  l_linestatus string,
  l_shipdate string,
  l_commitdate string,
  l_receiptdate string,
  l_shipinstruct string,
  l_shipmode string,
  l_comment string
)
STORED AS PARQUET
LOCATION 'oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/parquet_data/';

Criar uma tabela externa com base em um objeto TEXTFILE em um bucket do OSS

• Se os dados estiverem no formato JSON, armazenados como arquivos TEXTFILE em vários diretórios no OSS, e seguirem as mesmas regras de armazenamento e nomenclatura, use uma tabela particionada do MaxCompute para associá-los. A instrução DDL a seguir cria a tabela particionada:

  CREATE EXTERNAL TABLE tpch_lineitem_textfile
  (
    l_orderkey bigint,
    l_partkey bigint,
    l_suppkey bigint,
    l_linenumber bigint,
    l_quantity double,
    l_extendedprice double,
    l_discount double,
    l_tax double,
    l_returnflag string,
    l_linestatus string,
    l_shipdate string,
    l_commitdate string,
    l_receiptdate string,
    l_shipinstruct string,
    l_shipmode string,
    l_comment string
  )
  PARTITIONED BY (ds string)
  ROW FORMAT serde 'org.apache.hive.hcatalog.data.JsonSerDe'
  STORED AS TEXTFILE
  LOCATION 'oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data/';

• Os subdiretórios sob o diretório da tabela no OSS são organizados pelo nome da partição. Exemplo:

  oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data/ds=20170102/'
  oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data/ds=20170103/'
  ...

  Nesse caso, execute as seguintes instruções DDL para adicionar partições:

  ALTER TABLE tpch_lineitem_textfile ADD PARTITION(ds="20170102");
  ALTER TABLE tpch_lineitem_textfile ADD PARTITION(ds="20170103");

• Os subdiretórios sob o diretório da tabela no OSS não são organizados pelo nome da partição ou não estão no diretório da tabela. Exemplo:

  oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data_20170102/;
  oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data_20170103/;
  ...

  Nessa situação, execute as seguintes instruções DDL para adicionar partições:

  ALTER TABLE tpch_lineitem_textfile ADD PARTITION(ds="20170102")
  LOCATION 'oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data_20170102/';
  ALTER TABLE tpch_lineitem_textfile ADD PARTITION(ds="20170103")
  LOCATION 'oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data_20170103/';
  ...

• Não é possível personalizar a cláusula ROW FORMAT ao criar tabelas de dados TEXT. O código a seguir mostra o valor padrão da cláusula ROW FORMAT na instrução DDL:

  FIELDS TERMINATED BY : '\001'
  ESCAPED BY : '\'
  COLLECTION ITEMS TERMINATED BY : '\002'
  MAP KEYS TERMINATED BY : '\003'
  LINES TERMINATED BY : '\n'
  NULL DEFINED AS : '\N'

Criar uma tabela externa com base em um objeto CSV em um bucket do OSS

CREATE EXTERNAL TABLE [IF NOT EXISTS] 
(<column schemas>)
[PARTITIONED BY (partition column schemas)]
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.OpenCSVSerde'
  WITH SERDEPROPERTIES
    ('separatorChar'=',', 'quoteChar'='"', 'escapeChar'='\\')
STORED AS TEXTFILE
LOCATION 'oss://${endpoint}/${bucket}/${userfilePath}/';

separatorChar:','
quoteChar:'"'
escapeChar:'\'

Criar uma tabela externa com base em um objeto JSON em um bucket do OSS

CREATE EXTERNAL TABLE [IF NOT EXISTS] 
(<column schemas>)
[PARTITIONED BY (partition column schemas)]
ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe'
STORED AS TEXTFILE
LOCATION 'oss://${endpoint}/${bucket}/${userfilePath}/';

Criar uma tabela externa com base em um objeto ORC em um bucket do OSS

CREATE EXTERNAL TABLE [IF NOT EXISTS] 
(<column schemas>)
[PARTITIONED BY (partition column schemas)]
STORED AS ORC
LOCATION 'oss://${endpoint}/${bucket}/${userfilePath}/';

Criar uma tabela externa com base em um objeto AVRO em um bucket do OSS

CREATE EXTERNAL TABLE [IF NOT EXISTS] 
(<column schemas>)
[PARTITIONED BY (partition column schemas)]
STORED AS AVRO
LOCATION 'oss://${endpoint}/${bucket}/${userfilePath}/';

Criar uma tabela externa com base em um objeto SEQUENCEFILE em um bucket do OSS

CREATE EXTERNAL TABLE [IF NOT EXISTS] 
(<column schemas>)
[PARTITIONED BY (partition column schemas)]
STORED AS SEQUENCEFILE
LOCATION 'oss://${endpoint}/${bucket}/${userfilePath}/';

Ler e processar dados do OSS armazenados em formatos de código aberto

As instruções DDL anteriores mostram que basta modifique o valor após STORED AS para criar tabelas externas para diferentes formatos de objeto. Esta seção descreve como usar a tabela externa tpch_lineitem_parquet, criada com base em um objeto PARQUET. Para utilizar tabelas externas baseadas em objetos de outros formatos, basta definir STORED AS como PARQUET, ORC, TEXTFILE ou RCFILE.

• Leitura e processamento de dados do OSS em formatos de código aberto

  Após criar uma tabela externa e associá-la a dados específicos, gerencie a tabela criada como uma tabela padrão do MaxCompute.

  SELECT l_returnflag, l_linestatus,
  SUM(l_extendedprice*(1-l_discount)) AS sum_disc_price,
  AVG(l_quantity) AS avg_qty,
  COUNT(*) AS count_order
  FROM tpch_lineitem_parquet
  WHERE l_shipdate <= '1998-09-02'
  GROUP BY l_returnflag, l_linestatus;

  A tabela tpch_lineitem_parquet é usada como uma tabela interna. No entanto, o mecanismo de computação interna do MaxCompute lê diretamente os dados PARQUET do OSS para processamento.

  O parâmetro

  odps.sql.hive.compatible

  é definido como false se você usar as cláusulas

  ROW FORMAT

  e

  STORED AS

  para criar a tabela externa particionada tpch_lineitem_textfile. Para ler os dados corretamente, defina

  set odps.sql.hive.compatible

  como true. Caso contrário, um erro será retornado.

  SELECT * FROM tpch_lineitem_textfile LIMIT 1;
  FAILED: ODPS-0123131:User defined function exception - Traceback:
  com.aliyun.odps.udf.UDFException: java.lang.ClassNotFoundException: com.aliyun.odps.hive.wrapper.HiveStorageHandlerWrapper
  // Add the following flag that is compatible with Hive:
  set odps.sql.hive.compatible=true;
  SELECT * FROM tpch_lineitem_textfile LIMIT 1;
  +------------+------------+------------+--------------+------------+-----------------+------------+------------+--------------+--------------+------------+--------------+---------------+----------------+------------+-----------+
  | l_orderkey | l_partkey  | l_suppkey  | l_linenumber | l_quantity | l_extendedprice | l_discount | l_tax      | l_returnflag | l_linestatus | l_shipdate | l_commitdate | l_receiptdate | l_shipinstruct | l_shipmode | l_comment |
  +------------+------------+------------+--------------+------------+-----------------+------------+------------+--------------+--------------+------------+--------------+---------------+----------------+------------+-----------+
  | 5640000001 | 174458698  | 9458733    | 1            | 14.0       | 23071.58        | 0.08       | 0.06       | N            | O            | 1998-01-26 | 1997-11-16   | 1998-02-18    | TAKE BACK RETURN | SHIP       | cuses nag silently. quick |
  +------------+------------+------------+--------------+------------+-----------------+------------+------------+--------------+--------------+------------+--------------+---------------+----------------+------------+-----------+

  Nota

  • Ao usar uma tabela externa para leitura, cada operação aciona processos de I/O nos dados do OSS, impedindo que as otimizações de desempenho do MaxCompute para armazenamento interno tenham efeito. Consequentemente, o desempenho de leitura pode cair. Se você precisar de computação repetida de dados ou alta eficiência de cálculo, recomendamos importar os dados para o MaxCompute.

  • Quando instruções SQL utilizarem tipos de dados complexos, como CREATE, SELECT e INSERT, defina set odps.sql.type.system.odps2 como true antes das instruções. Em seguida, envie-as para execução. Para mais informações, consulte Guia de versão de tipos de dados.

• Importação de dados em formatos de código aberto para o MaxCompute visando computação

  Crie uma tabela interna do MaxCompute chamada tpch_lineitem_internal com o mesmo esquema da tabela externa. Importe os dados do OSS armazenados em formatos de código aberto para essa nova tabela interna, salvando-os no formato de armazenamento interno.

  CREATE TABLE tpch_lineitem_internal LIKE tpch_lineitem_parquet;
  INSERT OVERWRITE TABLE tpch_lineitem_internal;
  SELECT * FROM tpch_lineitem_parquet;

  Na tabela interna, execute instruções de consulta complexas baseadas na tabela externa para melhorar o desempenho da computação.

  SELECT l_returnflag, l_linestatus,
  SUM(l_extendedprice*(1-l_discount)) AS sum_disc_price,
  AVG(l_quantity) AS avg_qty,
  COUNT(*) AS count_order
  FROM tpch_lineitem_internal
  WHERE l_shipdate <= '1998-09-02'
  GROUP BY l_returnflag, l_linestatus;

Perguntas frequentes

Error: Inline data exceeds the maximum allowed size.

Cause: O armazenamento do OSS limita o tamanho de cada objeto. Se um objeto exceder 3 GB, um erro será relatado.

Solution: Ajuste os valores dos seguintes sinalizadores. Eles controlam o volume de dados que cada reducer pode gravar no armazenamento do OSS. Altere esses valores para garantir que o tamanho de um objeto armazenado no OSS não ultrapasse 3 GB.

set odps.sql.mapper.split.size=256; # Adjust the volume of table data that is read by each mapper. Unit: MB.
set odps.sql.reducer.instances=100; # Adjust the number of reducers in the execution plan.