MaxCompute:はじめに

データ準備

このトピックでは、MaxCompute のパブリックデータセットの maxframe_ml_100k_users テーブルを使用して、MaxFrame の使用方法を説明します。テストデータは、パブリック MaxCompute プロジェクト BIGDATA_PUBLIC_DATASET の data_science スキーマに保存されています。このデータは直接使用できます。

セッションの初期化

MaxFrame ジョブを実行する前に、MaxFrame セッションを初期化する必要があります。コードのエントリーポイントで、new_session API 操作を呼び出してジョブを初期化できます。その後のすべてのデータ処理は、作成されたセッションオブジェクトを介して実行され、バックエンドサービスと対話します。次のコードに例を示します。

import os
from maxframe import new_session
from odps import ODPS

# MaxFrame アカウントを使用して MaxCompute を初期化します。
o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='your-default-project',
    endpoint='your-end-point',
)

# MaxFrame セッションを初期化します。
new_session(odps_entry=o)

new_session API 操作の詳細については、「new_session」をご参照ください。

DataFrame オブジェクトの作成

read_odps_table および read_odps_query API 操作を使用すると、MaxCompute テーブルから DataFrame オブジェクトを作成できます。これらの DataFrame オブジェクトは、Pandas スタイルのデータ操作をサポートします。MaxFrame は、ローカルデータを使用した DataFrame オブジェクトの初期化もサポートしています。この機能は、コードを迅速にテストおよび開発するのに役立ちます。

• MaxCompute テーブルの使用

  read_odps_table API 操作を使用して、次の操作を実行できます。

  • read_odps_table 関数を使用して MaxCompute テーブルからデータを読み取り、それを MaxFrame DataFrame オブジェクトに変換できます。

    import maxframe.dataframe as md
    
    # MaxCompute テーブルからデータを読み取り、DataFrame オブジェクトを作成します。
    df = md.read_odps_table('BIGDATA_PUBLIC_DATASET.data_science.maxframe_ml_100k_users')

  • index_col パラメーターを設定して、テーブル内の列を DataFrame オブジェクトのインデックスとして指定します。

    import maxframe.dataframe as md
    
    # データベースの user_id 列を DataFrame オブジェクトのインデックスとして使用します。
    df = md.read_odps_table('BIGDATA_PUBLIC_DATASET.data_science.maxframe_ml_100k_users', index_col="user_id")

  • columns パラメーターを設定して、テーブル内の列を指定して DataFrame オブジェクトを作成します。

    import maxframe.dataframe as md
    
    # テーブル内の列を選択して DataFrame オブジェクトを作成します。
    df = md.read_odps_table('BIGDATA_PUBLIC_DATASET.data_science.maxframe_ml_100k_users', columns=['user_id', 'age', 'sex'])

    read_odps_table API の詳細については、「read_odps_table」をご参照ください。

• MaxCompute SQL のクエリ結果の使用

  テーブルから DataFrame を作成することに加えて、MaxFrame では read_odps_query API 操作で SQL クエリを実行し、その結果を DataFrame のデータ入力として使用できます。

  import maxframe.dataframe as md
  
  df = md.read_odps_query('select user_id, age, sex FROM `BIGDATA_PUBLIC_DATASET.data_science.maxframe_ml_100k_users`')

  index_col パラメーターを使用して、DataFrame オブジェクトのインデックスを指定します。

  import maxframe.dataframe as md
  
  df = md.read_odps_query('select user_id, age, sex FROM `BIGDATA_PUBLIC_DATASET.data_science.maxframe_ml_100k_users`', index_col='user_id')

  read_odps_query API の詳細については、「read_odps_query」をご参照ください。

• オンプレミスデータの使用

  MaxFrame は Pandas と同様の使用体験を提供します。オンプレミスデータに基づいて MaxFrame DataFrame オブジェクトを直接作成できます。サンプルコード:

  import maxframe.dataframe as md
  
  d = {'col1': [1, 2], 'col2': [3, 4]}
  df = md.DataFrame(data=d)
  print(df.execute().fetch())
  # 返された結果を取得します。
     col1  col2
  0     1     3
  1     2     4
  
  df = md.DataFrame(data=d, dtype=np.int8)
  print(df.execute().fetch())
  # 返された結果を取得します。
     col1  col2
  0     1     3
  1     2     4
  
  d = {'col1': [0, 1, 2, 3], 'col2': pd.Series([2, 3], index=[2, 3])}
  df = md.DataFrame(data=d, index=[0, 1, 2, 3])
  print(df.execute().fetch())
  # 返された結果を取得します。
     col1  col2
  0     0   NaN
  1     1   NaN
  2     2   2.0
  3     3   3.0
  
  data = np.array([(1, 2, 3), (4, 5, 6), (7, 8, 9)],
                  dtype=[("a", "i4"), ("b", "i4"), ("c", "i4")])
  df = md.DataFrame(data, columns=['c', 'a'])
  df.execute().fetch()
  # 返された結果を取得します。
     c  a
  0  3  1
  1  6  4
  2  9  7

  詳細については、「DataFrame」をご参照ください。

データ処理

MaxFrame は、Pandas と互換性のある一連の API を提供します。これらの API を使用して、データ計算、射影、フィルタリング、ソートなどのさまざまな操作を実行できます。MaxFrame は、一般的なデータ処理のための幅広いオペレーターもサポートしています。MaxFrame は、データ処理と分析の要件を満たすために UDF もサポートしています。

import maxframe.dataframe as md
df = md.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})
print(df.set_axis(['I', 'II'], axis='columns').execute().fetch())

# 返された結果を取得します。
   I  II 
0  1   4 
1  2   5 
2  3   6 

import maxframe.dataframe as md

df = md.DataFrame({'num_legs': [2, 4, 8, 0],
                        'num_wings': [2, 0, 0, 0],
                        'num_specimen_seen': [10, 2, 1, 8]},
                       index=['falcon', 'dog', 'spider', 'fish'])
print(df['num_legs'].sample(n=3, random_state=1).execute())

# 返された結果を取得します。
falcon    2
fish      0
dog       4
Name: num_legs, dtype: int64

import maxframe.dataframe as md
import numpy as np
df = md.DataFrame({                            
    'col1': ['A', 'A', 'B', np.nan, 'D', 'C'], 
    'col2': [2, 1, 9, 8, 7, 4],                
    'col3': [0, 1, 9, 4, 2, 3],                
})            
res = df.sort_values(by=['col1']).execute()
print(res.fetch())

# 返された結果を取得します。
   col1  col2  col3
0     A     2     0
1     A     1     1
2     B     9     9
5     C     4     3
4     D     7     2
3  None     8     4


res =  df.sort_values(by=['col1', 'col2']).execute()
print(res.fetch())

# 返された結果を取得します。
     col1 col2 col3
 1   A    1    1   
 0   A    2    0   
 2   B    9    9   
 5   C    4    3   
 4   D    7    2   
 3   None  8    4   

import maxframe.dataframe as md
df = md.DataFrame({'key': ['K0', 'K1', 'K2', 'K3', 'K4', 'K5'],
                        'A': ['A0', 'A1', 'A2', 'A3', 'A4', 'A5']})

other = md.DataFrame({'key': ['K0', 'K1', 'K2'],
                       'B': ['B0', 'B1', 'B2']})

print(df.join(other, lsuffix='_caller', rsuffix='_other').execute().fetch())

# 返された結果を取得します。
  key_caller   A key_other     B
0         K0  A0        K0    B0
1         K1  A1        K1    B1
2         K2  A2        K2    B2
3         K3  A3      None  None
4         K4  A4      None  None
5         K5  A5      None  None

import maxframe.dataframe as md

df = md.DataFrame([[1, 2, 3],
                   [4, 5, 6],
                   [7, 8, 9],
                   [np.nan, np.nan, np.nan]],
                  columns=['A', 'B', 'C'])
print(df.agg(['sum', 'min']).execute().fetch())

# 返された結果を取得します。
        A     B     C
min   1.0   2.0   3.0
sum  12.0  15.0  18.0

df.agg({'A' : ['sum', 'min'], 'B' : ['min', 'max']}).execute().fetch()

# 返された結果を取得します。
        A    B
max   NaN  8.0
min   1.0  2.0
sum  12.0  NaN

結果の保存

データセットが変換された後、to_odps_table API 操作を使用して結果を MaxCompute テーブルに保存できます。

• 例 1: 処理されたデータを MaxCompute テーブルに書き込みます。

  # 処理されたデータを MaxCompute テーブルに書き込みます。
  filtered_df.to_odps_table('<table_name>')

• 例 2: ストレージ期間を指定します。

  lifecycle パラメーターを設定して、結果テーブルのライフサイクルを指定できます。

  # 処理されたデータを MaxCompute テーブルに書き込みます。
  filtered_df.to_odps_table('<table_name>', lifecycle = 7)

パラメーター

table_name: 宛先の MaxCompute テーブルの名前。テーブルが MaxCompute に存在しない場合、テーブルは自動的に作成されます。テーブルが既に存在する場合、データはデフォルトで追加されます。

overwrite パラメーターを使用して、既存のデータを上書きするかどうかを指定できます。ストレージ操作の詳細については、「to_odps_table」をご参照ください。

タスクの実行と実行結果の表示

execute() メソッドを使用してデータ処理タスクをトリガーし、fetch() メソッドを使用して結果データの一部を取得できます。

# 実行結果の部分的なデータを取得します。
data = filtered_df.execute().fetch()
print(data)

タスクの実行と結果の取得の詳細については、「execute」および「fetch」をご参照ください。