Spark を使用した MaxCompute データへのアクセス - MaxCompute

MaxCompute のオープンストレージにより、Spark はコネクタを使用してストレージ API を呼び出し、MaxCompute から直接データを読み取ることができます。このアプローチにより、データ読み取りプロセスが簡素化され、アクセスパフォーマンスが向上します。Spark を MaxCompute のデータストレージと統合することで、効率的で柔軟、かつ強力なデータ処理および分析機能を提供します。

適用範囲

サードパーティのエンジンが MaxCompute にアクセスする場合：
- 標準テーブル、パーティションテーブル、クラスタ化テーブル、Delta テーブル、マテリアライズドビューからデータを読み取ることができます。
- MaxCompute の外部テーブルや論理ビューからデータを読み取ることはできません。
このコネクタは、JSON データ型の読み取りをサポートしていません。

操作手順

MaxCompute を有効化し、MaxCompute プロジェクトを作成します。
Data Transmission Service の専用リソースグループ (サブスクリプション) を購入。
Spark 開発環境をデプロイします。
Spark をクリックして、バージョン Spark 3.2.x - Spark 3.5.x の Spark パッケージをダウンロードし、ローカルフォルダーに解凍します。
1. Linux オペレーティングシステム で Spark 開発環境を構築するには、Linux 開発環境の構築をご参照ください。
2. Windows オペレーティングシステム で Spark 開発環境を構築するには、Windows 開発環境の構築をご参照ください。

Spark コネクタをダウンロードしてコンパイルします。現在、Spark バージョン 3.2.x から 3.5.x のみがサポートされています。このトピックでは、Spark 3.3.1 を例として使用します。

git clone コマンドを使用して Spark コネクターパッケージをダウンロードします。お使いの環境に Git がインストールされていることを確認してください。そうでない場合、コマンドを実行するとエラーが発生します。

## Spark コネクタをダウンロードします。
git clone https://github.com/aliyun/aliyun-maxcompute-data-collectors.git

## spark-connector フォルダーに移動します。
cd aliyun-maxcompute-data-collectors/spark-connector 

## コネクタをコンパイルします。
mvn clean package

## データソース JAR パッケージの場所。
datasource/target/spark-odps-datasource-3.3.1-odps0.43.0.jar

## データソース JAR パッケージを $SPARK_HOME/jars/ フォルダーにコピーします。
cp datasource/target/spark-odps-datasource-3.3.1-odps0.43.0.jar $SPARK_HOME/jars/

MaxCompute アカウントのアクセス情報を設定します。

Spark のインストール先にある conf フォルダーに、spark-defaults.conf ファイルを作成します：

cd $SPARK_HOME/conf
vim spark-defaults.conf

次のアカウント情報を spark-defaults.conf ファイルに追加します：

## spark-defaults.conf でアカウントを設定します。
spark.hadoop.odps.project.name=doc_test
spark.hadoop.odps.access.id=L********************
spark.hadoop.odps.access.key=*******************
spark.hadoop.odps.end.point=http://service.cn-beijing.maxcompute.aliyun.com/api
spark.hadoop.odps.tunnel.quota.name=ot_xxxx_p#ot_xxxx
## MaxCompute カタログを設定します。
spark.sql.catalog.odps=org.apache.spark.sql.execution.datasources.v2.odps.OdpsTableCatalog 
spark.sql.extensions=org.apache.spark.sql.execution.datasources.v2.odps.extension.OdpsExtensions

Spark コネクタを使用して MaxCompute にアクセスします。
1. Spark のインストール先にある bin フォルダーで次のコマンドを実行し、Spark SQL クライアントを起動します：
```
cd $SPARK_HOME/bin
spark-sql
```
2. MaxCompute プロジェクト内のテーブルを照会します：
```
SHOW tables in odps.doc_test;
```
  doc_test は MaxCompute プロジェクト名の一例です。実際の MaxCompute プロジェクト名に置き換えてください。
3. テーブルを作成します：
```
CREATE TABLE odps.doc_test.mc_test_table (name STRING, num BIGINT);
```
4. テーブルからデータを読み取ります：
```
SELECT * FROM odps.doc_test.mc_test_table;
```
5. パーティションテーブルを作成します：
```
 CREATE TABLE odps.doc_test.mc_test_table_pt (name STRING, num BIGINT) PARTITIONED BY (pt1 STRING, pt2 STRING);
```
6. パーティションテーブルからデータを読み取ります：
```
SELECT * FROM odps.doc_test.mc_test_table_pt;
```
  次の出力が返されます：
```
test1   1       2018    0601
test2   2       2018    0601
Time taken: 1.312 seconds, Fetched 2 row(s)
```
7. テーブルを削除します：
```
DROP TABLE IF EXISTS odps.doc_test.mc_test_table;
```