オープンソース形式で保存されたOSSデータの処理 - - Alibaba Cloud ドキュメントセンター

このトピックでは、MaxCompute非構造化フレームワークを使用して、オープンソース形式で保存されているObject Storage Service (OSS) データを処理する方法について説明します。フォーマットは、ORC、PARQUET、SEQUENCEFILE、RCFILE、AVRO、およびTEXTFILEを含む。

DataWorksコンソールで外部テーブルを作成、検索、設定、および処理できます。外部テーブルのデータを照会および分析することもできます。詳細については、「外部テーブル」をご参照ください。

オープンソース形式で格納されているデータの場合、非構造化フレームワークは、オープンソースコミュニティが提供する実装メソッドを呼び出してデータを解析します。このメソッドをMaxComputeとシームレスに統合して、オープンソース形式で保存されたデータを読み取り、解析できます。

説明

オープンソース形式で保存されたOSSデータを処理する前に、まずSTSを使用してOSSへのアクセスを許可する必要があります。詳細については、「STS権限付与」をご参照ください。

外部テーブルを作成する構文

MaxCompute非構造化フレームワークは、外部テーブルを使用してさまざまな形式で保存されたデータにアクセスします。次のコードでは、外部テーブルの作成に使用される構文について説明します。次に、テーブルを使用して、オープンソース形式で保存されているOSSデータにアクセスできます。

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}/';

構文はHiveの構文と似ています。次の点に注意してください。

column schemas: 外部テーブルの列を定義します。列の定義は、OSSに保存されているデータの定義と同じである必要があります。
ROW FORMAT SERDE: この句は、TEXTFILEなどの特殊な形式を使用する場合にのみ必要です。
WITH SERDEPROPERTIES: OSSでSTS権限付与を行う場合、この句はodps.properties.rolearnプロパティを設定するために必要です。プロパティは、RAMロールのAlibaba Cloudリソース名 (ARN) です。ファイル形式を指定するには、STORED AS <file format> と <serde class> を設定します。次の例では、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}/'
```
説明
STS権限付与が実行されない場合、odps.properties.rolearnプロパティを設定する必要はありません。 Location句のAccessKey IDとAccessKey secretをプレーンテキストで指定するだけです。
STORED AS: オープンソース形式で保存されているデータの読み取りに固有です。これは、標準の非構造化外部テーブルの作成に使用されるSTORED BY句とは異なります。
STORED ASの後には、ORC、PARQUET、RCFILE、SEQUENCEFILE、またはTEXTFILEなどのファイルフォーマットが続く。 STORED AS句では、ファイルのサイズは3 GBを超えることはできません。ファイルのサイズが3 GBを超える場合は、ファイルを分割します。
ファイル形式とSerDeクラス間のマッピング:
- シーケンス: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
- TEXTFILE: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
- RCFILE: org.apache.hadoop.hive. serde2.columnary. 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: OSSサービスが関連付けられている場合、AccessKey IDとAccessKeyシークレットをプレーンテキストで設定します。例：
```
LOCATION 'oss://${accessKeyId}:${accessKeySecret}@${endpoint}/${bucket}/${userPath}/'
```
外部テーブルを使用して、OSSのパブリックエンドポイントを使用してOSSデータにアクセスすることはできません。

OSSバケット内のPARQUETオブジェクトに基づいて外部テーブルを作成する

一部のオブジェクトがOSSディレクトリに格納され、これらのオブジェクトはPARQUET形式で格納され、各オブジェクトには16個の列が含まれています。4つのBIGINTタイプの列、4つのDOUBLEタイプの列、8つのSTRINGタイプの列です。次のDDLステートメントを使用して、テーブルを作成します。

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/';

PARQUET形式で保存されたデータは、デフォルトでは圧縮されません。 MaxComputeでこのようなデータを圧縮する場合は、se t odps.sql.hive.com patibleをtrueに設定します。圧縮後のデータの形式は、TBLPROPERTIESパラメーターの値によって異なります。

'mcfed .parquet.com pression'='SNAPPY': 形式はSNAPPYです。
'mcfed .parquet.com pression'='GZIP' 形式はGZIPです。

OSSバケット内のTEXTFILEオブジェクトに基づいて外部テーブルを作成する

データがJSON形式であり、OSSの複数のディレクトリにTEXTFILEファイルとして保存され、ファイルストレージと命名が同じ規則に準拠している場合、MaxComputeパーティションテーブルを使用してデータを関連付けることができます。次のDDLステートメントを使用して、パーティションテーブルを作成します。

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/';

OSSテーブルディレクトリの下のサブディレクトリは、パーティション名で整理されています。例：

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/'
...

この場合、次のDDL文を実行してパーティションを追加します。

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

OSSテーブルディレクトリの下のサブディレクトリは、パーティション名で整理されていません。例：

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/;
...

この場合、次のDDL文を実行してパーティションを追加します。

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/';
...

TEXTデータテーブルを作成するときに、ROW FORMAT句をカスタマイズすることはできません。次のコードは、DDLステートメントのROW FORMAT句のデフォルト値を示しています。
```
FIELDS TERMINATED BY : '\001'
ESCAPED BY : '\'
COLLECTION ITEMS TERMINATED BY : '\002'
MAP KEYS TERMINATED BY : '\003'
LINES TERMINATED BY : '\n'
NULL DEFINED AS : '\N'
```

OSSバケット内のCSVオブジェクトに基づいて外部テーブルを作成する

次のDDLステートメントを使用して、外部テーブルを作成します。

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}/';

次のコードは、DDLステートメントのWITH SERDEPROPERTIES句のフィールドとそのデフォルト値を説明しています。

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

説明

Hive OpenCSVSerdeはSTRINGタイプのみをサポートします。 OpenCSVSerdeは組み込みのSerDeではありません。 DMLステートメントを実行するには、se t odps.sql.hive.com patibleをtrueに設定する必要があります。

OSSバケット内のJSONオブジェクトに基づいて外部テーブルを作成する

WITH SERDEPROPERTIES句をサポートする次のDDLステートメントを使用して、外部テーブルを作成します。

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}/';

OSSバケット内のORCオブジェクトに基づいて外部テーブルを作成する

次のDDLステートメントを使用して、外部テーブルを作成します。

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

OSSバケット内のAVROオブジェクトに基づいて外部テーブルを作成する

次のDDLステートメントを使用して、外部テーブルを作成します。

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

OSSバケット内のSEQUENCEFILEオブジェクトに基づいて外部テーブルを作成する

次のDDLステートメントを使用して、外部テーブルを作成します。

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

オープンソース形式で保存されたOSSデータの読み取りと処理

上記のDDLステートメントは、異なるオブジェクト形式の外部テーブルを作成するには、STORED ASの後に値を変更するだけでよいことを示しています。このセクションでは、PARQUETオブジェクトに基づいて作成されるtpch_lineitem_parquet外部テーブルの使用方法について説明します。さまざまな形式のオブジェクトに基づいて作成された外部テーブルを使用するには、STORED ASをPARQUET、ORC、TEXTFILE、またはRCFILEに設定するだけです。

オープンソース形式で保存されたOSSデータの読み取りと処理

外部テーブルを作成して特定のデータに関連付けると、作成したテーブルを標準の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;

tpch_lineitem_parquetは内部テーブルとして使用されます。ただし、MaxCompute内部コンピューティングエンジンは、処理のためにOSSからPARQUETデータを直接読み取ります。

ROW FORMATおよびSTORED AS句を使用してtpch_lineitem_textfile外部パーティションテーブルを作成する場合、odps.sql.hive.com patibleはfalseに設定されます。データを正しく読み取るには、se t odps.sql.hive.com patibleをtrueに設定する必要があります。それ以外の場合は、エラーが返されます。

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 |
+------------+------------+------------+--------------+------------+-----------------+------------+------------+--------------+--------------+------------+--------------+---------------+----------------+------------+-----------+

説明

外部テーブルを使用してデータを読み取る場合、各データ読み取り操作はOSSデータに対するI/O操作をトリガーし、内部ストレージのMaxComputeパフォーマンス最適化は有効になりません。その結果、データの読み出し性能が低下するおそれがある。反復データコンピューティングまたは高いコンピューティング効率が必要な場合は、MaxComputeにデータをインポートしてコンピューティングを行うことを推奨します。
SQLステートメントでCREATE、SELECT、INSERTなどの複雑なデータ型が使用されている場合は、ステートメントの前にset odps.sql.type.system.odps2をtrueに設定する必要があります。次に、実行のためにステートメントをコミットします。詳細については、「Data type editions」をご参照ください。

オープンソース形式で保存されたデータをMaxComputeにインポートしてコンピューティング
外部テーブルと同じスキーマを持つ、tpch_lineitem_internalという名前のMaxCompute内部テーブルを作成します。オープンソース形式で保存されたOSSデータを新しく作成した内部テーブルにインポートします。内部ストレージ形式でデータを保存します。
```
CREATE TABLE tpch_lineitem_internal LIKE tpch_lineitem_parquet;
INSERT OVERWRITE TABLE tpch_lineitem_internal;
SELECT * FROM tpch_lineitem_parquet;
```
内部テーブルで、外部テーブルに基づいて複雑なクエリ文を実行し、コンピューティングパフォーマンスを向上させます。
```
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;
```

よくある質問

エラー: インラインデータが最大許容サイズを超えています。

原因: OSSストレージは各オブジェクトのサイズを制限します。オブジェクトのサイズが3 GBを超えると、エラーが報告されます。

解決策: 次のフラグの値を調整します。フラグは、各レデューサーがOSSストレージに書き込むことができるデータの量を制御します。フラグの値を変更して、OSSに保存されているオブジェクトのサイズが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.