PyODPS读取分区表数据 - MaxCompute - Alibaba Cloud ドキュメントセンター

この記事では、PyODPS を使用してパーティションテーブルのデータを読み取る方法について説明します。

前提条件

以下の操作を完了している必要があります。

MaxCompute が有効化されていること。
DataWorks が有効化されていること。
DataWorks で業務フローが作成されていること。詳細については、「ワークフローの作成」をご参照ください。

操作手順

説明

この例では DataWorks のシンプルモードを使用します。ワークスペースの作成時に、デフォルトで [Data Studio のパブリックベータ版に参加] は無効になっています。パブリックベータ版のワークスペースは、この例には適用されません。

テストデータを準備します。
1. テーブルを作成し、データをアップロードします。操作方法については、「Create a table and upload data」をご参照ください。
  テーブル構造とソースデータの情報は次のとおりです：
  - パーティションテーブル user_detail のテーブル作成文は次のとおりです。
```
CREATE TABLE IF NOT EXISTS user_detail
(
userid    BIGINT COMMENT 'ユーザー ID',
job       STRING COMMENT '職種',
education STRING COMMENT '学歴'
) COMMENT 'ユーザー情報テーブル'
PARTITIONED BY (dt STRING COMMENT '日付', region STRING COMMENT 'リージョン');
```
  - ソースデータテーブル user_detail_ods のテーブル作成文は以下のとおりです。
```
CREATE TABLE IF NOT EXISTS user_detail_ods
(
  userid    BIGINT COMMENT 'ユーザー ID',
  job       STRING COMMENT '職種',
  education STRING COMMENT '学歴',
  dt STRING COMMENT '日付',
  region STRING COMMENT 'リージョン'
);
```
  - テストデータを user_detail.txt ファイルとして保存し、テーブル user_detail_ods にアップロードします。
```
0001,互联网,本科,20190715,beijing
0002,教育,大专,20190716,beijing
0003,金融,硕士,20190715,shandong
0004,互联网,硕士,20190715,beijing
```
2. ソースデータテーブル user_detail_ods のデータをパーティションテーブル user_detail に書き込みます。
  1. DataWorks コンソールにログインします。
  2. 左側メニューで、[Workspace] をクリックします。
  3. 対象のワークスペースを確認し、[操作] 列の 快速进入 > データ開発 を選択します。
  4. ワークフローを右クリックし、Create Node > ODPS SQL を選択します。
  5. ノード名を入力し、[確認] をクリックします。
  6. ODPS SQL ノードに次のコードを入力します。
```
INSERT OVERWRITE TABLE user_detail PARTITION (dt, region) 
SELECT userid, job, education, dt, region FROM user_detail_ods;
```
  7. [実行] をクリックして、データの書き込みを完了します。

PyODPS によるパーティションテーブルデータの読み取り

DataWorks コンソールにログインします。
左側メニューで、[Workspace] をクリックします。
対象のワークスペースを確認し、[操作] 列の 快速进入 > データ開発 を選択します。
[データ開発] ページで、作成済みのワークフローを右クリックし、Create Node > PyODPS 2 を選択します。
ノード名を入力し、[確認] をクリックします。

PyODPS 2 ノードに次のコードを入力して、パーティションテーブルのデータを読み取ります。

import sys
from odps import ODPS
reload(sys)
print('dt=' + args['dt'])
# システムのデフォルトエンコーディングを utf8 に変更します。
sys.setdefaultencoding('utf8')
# テーブル 'user_detail' のオブジェクトを取得します。
t = o.get_table('user_detail')
# パーティションが存在するかどうかを確認します。
print t.exist_partition('dt=20190715,region=beijing')
# すべてのパーティションを表示します。
for partition in t.partitions:
    print partition.name
# 以下の 3 つの方法でパーティションテーブルのデータを照会できます。
# 1 番目の方法は次のとおりです。
print("1番目の方法でパーティションテーブルのデータを照会:")
with t.open_reader(partition='dt=20190715,region=beijing') as reader1:
    count = reader1.count
    for record in reader1:
        print record[0],record[1],record[2]
# 2 番目の方法は次のとおりです。
print("2番目の方法でパーティションテーブルのデータを照会:")
reader2 = t.open_reader(partition='dt=20190715,region=beijing')
for record in reader2:
    print record["userid"],record["job"],record["education"]
# 3 番目の方法は次のとおりです。
print("3番目の方法でパーティションテーブルのデータを照会:")
for record in o.read_table('user_detail', partition='dt=20190715,region=beijing'):
    print record["userid"],record["job"],record["education"]

Run with Parameters をクリックします。
[Parameters] ダイアログボックスに設定パラメーターを入力し、[実行] をクリックします。
設定パラメーターの説明は次のとおりです：
- [リソースグループ名]： [Shared Resource Group] を選択します。
- [dt]： 20190715 に設定します。

Runtime Log で実行結果を確認します。

Executing user script with PyODPS 0.8.0
dt=20190715
True
dt='20190715',region='beijing'
dt='20190715',region='shandong'
dt='20190716',region='beijing'
1番目の方法でパーティションテーブルのデータを照会:
4 インターネット 修士
1 インターネット 大学卒
2番目の方法でパーティションテーブルのデータを照会:
4 インターネット 修士
1 インターネット 大学卒
3番目の方法でパーティションテーブルのデータを照会:
4 インターネット 修士
1 インターネット 大学卒