Jupyterlab ノートブックでのデータアクセス

サポートされる各カーネルは、ノートブック内のデータセットから Platform データを読み取るための組み込み機能を備えてます。現在、Adobe Experience Platform Data Science Workspace の JupyterLab は Python、R、PySpark、Scala のノートブックをサポートしています。 ただし、ページ番号付けデータのサポートは、Python と R ノートブックに限られます。 このガイドでは、JupyterLab ノートブックを使用してデータにアクセスする方法に焦点を当てています。

はじめに

このガイドを読む前に、JupyterLab ユーザーガイド を参照し、JupyterLab の概要と Data Science Workspace 内での役割について確認してください。

ノートブックデータの制限

次の情報は、読み取り可能な最大データ量、使用されたデータのタイプ、およびデータを読み取る推定時間枠を定義します。

Python と R では、40 GB の RAM で構成されたノートブックサーバがベンチマークに使用されました。 PySpark と Scala の場合、64GB の RAM、8 コア、2 DBU で構成され、最大 4 人のワーカーを持つデータベースクラスタが以下のベンチマークに使用されました。

使用される ExperienceEvent スキーマデータのサイズは、1,000 行から 10 億行 (1B) までの範囲で様々です。 PySpark と Spark 指標の場合、XDM データには 10 日間の日付範囲が使用されていました。

アドホックスキーマデータは、Query Service Create Table as Select (CTAS) を使用して事前に処理されています。 このデータは、1,000 行 (1K) から 10 億行 (1B) までのサイズも異なります。

バッチモードとインタラクティブモードを使用するタイミング

PySpark と Scala のノートブックでデータセットを読み取る場合、インタラクティブモードまたはバッチモードを使用してデータセットを読み取ることができます。 インタラクティブは高速な結果を得るために作成されるのに対して、バッチモードは大規模なデータセット用です。

• PySpark と Scala のノートブックの場合、500 万行以上のデータを読み取る場合は、バッチモードを使用する必要があります。 各モードの効率について詳しくは、以下の PySpark または Scala データ制限テーブルを参照してください。

Python ノートブックデータの制限

XDM ExperienceEvent スキーマ： XDM データの最大 200 万行（ディスク上の約 6.1 GB のデータ）を 22 分以内に読み取ることができます。行を追加すると、エラーが発生する場合があります。

アドホックスキーマ： 非 XDM（アドホック）データの最大 500 万行（ディスク上の約 5.6 GB のデータ）を 14 分以内に読み取ることができます。行を追加すると、エラーが発生する場合があります。

R ノートブックのデータの制限

XDM ExperienceEvent スキーマ： 13 分未満で、最大 100 万行の XDM データ（ディスク上の 3 GB データ）を読み取ることができます。

アドホックスキーマ： アドホックデータの最大 300 万行（ディスク上のデータは 293 MB）を約 10 分で読み取れるはずです。

PySpark（Python カーネル）ノートブックのデータ制限：

XDM ExperienceEvent スキーマ： インタラクティブモードでは、XDM データの最大 500 万行（ディスク上の約 13.42 GB のデータ）を約 20 分で読み取ることができます。インタラクティブモードでは、最大 500 万行までサポートされます。 大きなデータセットを読み取る場合は、バッチモードに切り替えることをお勧めします。 バッチモードでは、約 14 時間で XDM データの最大 5 億行（ディスク上の最大 1.31 TB のデータ）を読み取れるはずです。

アドホックスキーマ： インタラクティブモードでは、XDM 以外のデータの最大 500 万行（ディスク上の約 5.36 GB のデータ）を 3 分以内に読み取ることができます。バッチモードでは、XDM 以外のデータの最大 10 億行（ディスク上の最大 1.05 TB のデータ）を約 18 分で読み取ることができます。

Spark （Scala カーネル）ノートブックのデータ制限：

XDM ExperienceEvent スキーマ： インタラクティブモードでは、XDM データの最大 500 万行（ディスク上の約 13.42 GB のデータ）を約 18 分で読み取ることができます。インタラクティブモードでは、最大 500 万行までサポートされます。 大きなデータセットを読み取る場合は、バッチモードに切り替えることをお勧めします。 バッチモードでは、約 14 時間で XDM データの最大 5 億行（ディスク上の最大 1.31 TB のデータ）を読み取れるはずです。

アドホックスキーマ： インタラクティブモードでは、XDM 以外のデータの最大 500 万行（ディスク上の約 5.36 GB のデータ）を 3 分以内に読み取ることができます。バッチモードでは、XDM 以外のデータの最大 10 億行（ディスク上のデータは約 1.05 TB）を約 16 分で読み取ることができます。

Python ノートブック

Python ノートブックを使用すると、データセットにアクセスする際にデータをページ番号付けできます。ページ番号付けの有無に関わらずデータを読み取るコード例を以下に示します。利用可能なスターター Python ノートブックの詳細については、JupyterLab ユーザーガイド内の JupyterLab Launcher の節を参照してください。

以下の Python ドキュメントでは、次の概念の概要を説明しています。

• データセットからの読み取り
• データセットへの書き込み
• クエリデータ
• ExperienceEvent データのフィルター

Python でのデータセットからの読み取り

ページ番号なし：

次のコードを実行すると、データセット全体が読み取られます。実行が成功した場合、データは df 変数で参照される Pandas データフレームとして保存されます。

# Python

from platform_sdk.dataset_reader import DatasetReader
dataset_reader = DatasetReader(get_platform_sdk_client_context(), dataset_id="{DATASET_ID}")
df = dataset_reader.read()
df.head()

ページ番号付き：

次のコードを実行すると、指定したデータセットからデータが読み取られます。ページ番号付けは、limit() 関数を使用してデータを制限し、offset() 関数を使用してデータをオフセットすることで実現します。データの制限とは、読み取るデータポイントの最大数を指し、オフセットとは、データの読み取り前にスキップするデータポイントの数を指します。読み取り操作が正常に実行された場合、データは df 変数が参照する Pandas データフレームとして保存されます。

# Python

from platform_sdk.dataset_reader import DatasetReader

dataset_reader = DatasetReader(get_platform_sdk_client_context(), dataset_id="{DATASET_ID}")
df = dataset_reader.limit(100).offset(10).read()

Python でのデータセットへの書き込み

JupyterLab ノートブックのデータセットに書き込むには、JupyterLab の左側のナビゲーションで「データアイコン」タブ（下でハイライト表示）を選択します。 Datasets ディレクトリと Schemas ディレクトリが表示されます。 「データセット」を選択して右クリックし、使用するデータセットのドロップダウンメニューから「ノートブックにデータを書き込む」オプションを選択します。 ノートブックの下部に実行可能コードエントリが表示されます。

• 「Write Data in Notebook」を使用して、選択したデータセットで書き込みセルを生成します。
• Explore Data in Notebook を使用して、選択したデータセットで読み取りセルを生成します。
• ノートブックのクエリデータ を使用して、選択したデータセットで基本的なクエリセルを生成します。

または、次のコードセルをコピーして貼り付けることもできます。 {DATASET_ID} と {PANDA_DATAFRAME} の両方を置き換えます。

from platform_sdk.models import Dataset
from platform_sdk.dataset_writer import DatasetWriter

dataset = Dataset(get_platform_sdk_client_context()).get_by_id(dataset_id="{DATASET_ID}")
dataset_writer = DatasetWriter(get_platform_sdk_client_context(), dataset)
write_tracker = dataset_writer.write({PANDA_DATAFRAME}, file_format='json')

Python の Query Service を使用してデータをクエリする

JupyterLabPlatformPython で を使用すると、 ノートブックで SQL を使用して、Adobe Experience Platform クエリサービスを通じてデータにアクセスできます。Query Service を通じたデータへのアクセスは、実行時間が優れているので、大規模なデータセットを扱う場合に便利です。 Query Service を使用したデータのクエリには、処理時間の制限は 10 分です。

JupyterLab で Query Service を使用する前に、Query Service SQL 構文 に関する十分な知識があることを確認してください。

Query Service を使用してデータをクエリする場合は、ターゲットデータセットの名前を指定する必要があります。 必要なコードセルを生成するには、データエクスプローラー​を使用して目的のデータセットを見つけます。データセットの一覧を右クリックし、「ノートブック内のデータのクエリ」をクリックして、ノートブックに 2 つのコードセルを生成します。 これら 2 つのセルの詳細を以下に示します。

JupyterLab で Query Service を利用するには、まず作業中の Python ノートブックと Query Service の間に接続を作成する必要があります。 これは、最初に生成されたセルを実行することで達成できます。

qs_connect()

2 番目に生成されたセルでは、最初の行を SQL クエリの前に定義する必要があります。デフォルトでは、生成されたセルは、クエリ結果を Pandas データフレームとして保存するオプションの変数（df0）を定義します。
引数 -c QS_CONNECTION は必須で、カーネルに対して SQL クエリを実行するように指示しま Query Serviceす。その他の引数のリストは、付録を参照してください。

%%read_sql df0 -c QS_CONNECTION
SELECT *
FROM name_of_the_dataset
LIMIT 10
/* Querying table "name_of_the_dataset" (datasetId: {DATASET_ID})*/

Python 変数は、次の例に示すように、文字列形式の構文を使用して中括弧（{}）で囲むことで、SQL クエリ内で直接参照できます。

table_name = 'name_of_the_dataset'
table_columns = ','.join(['col_1','col_2','col_3'])

%%read_sql demo -c QS_CONNECTION
SELECT {table_columns}
FROM {table_name}

ExperienceEvent データをフィルタ

Python ノートブックの ExperienceEvent データセットにアクセスしてフィルタリングするには、データセットの ID({DATASET_ID}) と、論理演算子を使用して特定の時間範囲を定義するフィルタールールを指定する必要があります。 時間範囲を定義すると、指定されたページ番号は無視され、データセット全体が考慮されます。

フィルタリング操作のリストを以下に示します。

• eq()：と等しい
• gt()：より大きい
• ge()：より大きいか等しい
• lt()：より小さい
• le()：より小さいか等しい
• And()：論理積演算子
• Or()：論理和演算子

次のセルでは、ExperienceEvent データセットを、2019 年 1 月 1 日から 2019 年 12 月 31 日の終わりまでの間にのみ存在するデータにフィルタリングします。

# Python

from platform_sdk.dataset_reader import DatasetReader

dataset_reader = DatasetReader(get_platform_sdk_client_context(), dataset_id="{DATASET_ID}")
df = dataset_reader.\
    where(dataset_reader["timestamp"].gt("2019-01-01 00:00:00").\
    And(dataset_reader["timestamp"].lt("2019-12-31 23:59:59"))\
).read()

R ノートブック

R ノートブックを使用すると、データセットにアクセスする際にデータをページ番号付けできます。 ページ番号付けの有無に関わらずデータを読み取るコード例を以下に示します。使用可能なスターター R ノートブックの詳細については、JupyterLab ユーザーガイド内の JupyterLab Launcher の節を参照してください。

以下の R ドキュメントでは、次の概念の概要を説明します。

• データセットからの読み取り
• データセットへの書き込み
• ExperienceEvent データのフィルター

R でのデータセットからの読み取り

ページ番号なし：

次のコードを実行すると、データセット全体が読み取られます。実行が成功した場合、データは df0 変数で参照される Pandas データフレームとして保存されます。

# R

library(reticulate)
use_python("/usr/local/bin/ipython")
psdk <- import("platform_sdk")
datetime <- import("datetime", convert = FALSE)
py_run_file("~/.ipython/profile_default/startup/platform_sdk_context.py")
DatasetReader <- psdk$dataset_reader$DatasetReader
dataset_reader <- DatasetReader(py$get_platform_sdk_client_context(), dataset_id="{DATASET_ID}")
df0 <- dataset_reader$read()
head(df0)

ページ番号付き：

次のコードを実行すると、指定したデータセットからデータが読み取られます。ページ番号付けは、limit() 関数を使用してデータを制限し、offset() 関数を使用してデータをオフセットすることで実現します。データの制限とは、読み取るデータポイントの最大数を指し、オフセットとは、データの読み取り前にスキップするデータポイントの数を指します。読み取り操作が正常に実行された場合、データは df0 変数が参照する Pandas データフレームとして保存されます。

# R

library(reticulate)
use_python("/usr/local/bin/ipython")
psdk <- import("platform_sdk")
datetime <- import("datetime", convert = FALSE)
py_run_file("~/.ipython/profile_default/startup/platform_sdk_context.py")

DatasetReader <- psdk$dataset_reader$DatasetReader
dataset_reader <- DatasetReader(py$get_platform_sdk_client_context(), dataset_id="{DATASET_ID}") 
df0 <- dataset_reader$limit(100L)$offset(10L)$read()

R でのデータセットへの書き込み

JupyterLab ノートブックのデータセットに書き込むには、JupyterLab の左側のナビゲーションで「データアイコン」タブ（下でハイライト表示）を選択します。 Datasets ディレクトリと Schemas ディレクトリが表示されます。 「データセット」を選択して右クリックし、使用するデータセットのドロップダウンメニューから「ノートブックにデータを書き込む」オプションを選択します。 ノートブックの下部に実行可能コードエントリが表示されます。

• 「Write Data in Notebook」を使用して、選択したデータセットで書き込みセルを生成します。
• Explore Data in Notebook を使用して、選択したデータセットで読み取りセルを生成します。

または、次のコードセルをコピーして貼り付けることもできます。

psdk <- import("platform_sdk")
dataset <- psdk$models$Dataset(py$get_platform_sdk_client_context())$get_by_id(dataset_id="{DATASET_ID}")
dataset_writer <- psdk$dataset_writer$DatasetWriter(py$get_platform_sdk_client_context(), dataset)
write_tracker <- dataset_writer$write(df, file_format='json')

ExperienceEvent データをフィルタ

R ノートブック内の ExperienceEvent データセットにアクセスしてフィルタリングするには、データセットの ID({DATASET_ID}) と、論理演算子を使用して特定の時間範囲を定義するフィルタールールを指定する必要があります。 時間範囲を定義すると、指定されたページ番号は無視され、データセット全体が考慮されます。

フィルタリング操作のリストを以下に示します。

• eq()：と等しい
• gt()：より大きい
• ge()：より大きいか等しい
• lt()：より小さい
• le()：より小さいか等しい
• And()：論理積演算子
• Or()：論理和演算子

次のセルでは、ExperienceEvent データセットを、2019 年 1 月 1 日から 2019 年 12 月 31 日の終わりまでの間にのみ存在するデータにフィルタリングします。

# R

library(reticulate)
use_python("/usr/local/bin/ipython")
psdk <- import("platform_sdk")
datetime <- import("datetime", convert = FALSE)
py_run_file("~/.ipython/profile_default/startup/platform_sdk_context.py")

client_context <- py$PLATFORM_SDK_CLIENT_CONTEXT
DatasetReader <- psdk$dataset_reader$DatasetReader
dataset_reader <- DatasetReader(py$get_platform_sdk_client_context(), dataset_id="{DATASET_ID}") 

df0 <- dataset_reader$
    where(dataset_reader["timestamp"]$gt("2019-01-01 00:00:00")$
    And(dataset_reader["timestamp"]$lt("2019-12-31 23:59:59"))
)$read()

PySpark 3 ノートブック

以下の PySpark ドキュメントでは、次の概念の概要を説明しています。

• sparkSession の初期化
• データの読み取りと書き込み
• ローカルデータフレームの作成
• ExperienceEvent データのフィルター

sparkSession の初期化

すべての Spark 2.4 ノートブックでは、次のテンプレートコードを使用してセッションを初期化する必要があります。

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

%dataset を使用した PySpark 3 ノートブックでの読み書き

Spark 2.4 の導入に伴い、PySpark 3(Spark 2.4) ノートブックで使用できるように %dataset カスタムマジックが提供されました。 IPython カーネルで利用可能なマジックコマンドの詳細については、IPython マジックのドキュメント を参照してください。

用途

%dataset {action} --datasetId {id} --dataFrame {df} --mode batch

説明

PySpark ノートブック（Python 3 カーネル）からデータセットを読み書きするためのカスタム Data Science Workspace マジックコマンド。

例

• 読み取り例: %dataset read --datasetId 5e68141134492718af974841 --dataFrame pd0 --mode batch
• 記述例: %dataset write --datasetId 5e68141134492718af974842 --dataFrame pd0 --mode batch

次の方法を使用して、JupyterLab buy で上記の例を自動生成できます。

JupyterLab の左側のナビゲーションで、「データアイコン」タブ（下でハイライト表示）を選択します。 Datasets ディレクトリと Schemas ディレクトリが表示されます。 「データセット」を選択して右クリックし、使用するデータセットのドロップダウンメニューから「ノートブックにデータを書き込む」オプションを選択します。 ノートブックの下部に実行可能コードエントリが表示されます。

• 「Explore Data in Notebook」 を使用して、読み取りセルを生成します。
• 「Write Data in Notebook」を使用して、書き込みセルを生成します。

ローカルデータフレームの作成

PySpark 3 を使用してローカルデータフレームを作成するには、SQL クエリを使用します。 以下に例を示します。

date_aggregation.createOrReplaceTempView("temp_df")

df = spark.sql('''
  SELECT *
  FROM sparkdf
''')

local_df

df = spark.sql('''
  SELECT *
  FROM sparkdf
  LIMIT limit
''')

sample_df = df.sample(fraction)

ExperienceEvent データをフィルタ

PySpark ノートブック内の ExperienceEvent データセットにアクセスしてフィルタリングするには、データセット ID({DATASET_ID})、組織の IMS ID、および特定の時間範囲を定義するフィルタールールを指定する必要があります。 フィルタ時間範囲は、関数 spark.sql() を使用して定義します。関数パラメータは SQL クエリ文字列です。

次のセルでは、ExperienceEvent データセットを、2019 年 1 月 1 日から 2019 年 12 月 31 日の終わりまでの間にのみ存在するデータにフィルタリングします。

# PySpark 3 (Spark 2.4)

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

%dataset read --datasetId {DATASET_ID} --dataFrame df --mode batch

df.createOrReplaceTempView("event")
timepd = spark.sql("""
    SELECT *
    FROM event
    WHERE timestamp > CAST('2019-01-01 00:00:00.0' AS TIMESTAMP)
    AND timestamp < CAST('2019-12-31 23:59:59.9' AS TIMESTAMP)
""")
timepd.show()

Scala ノートブック

以下のドキュメントには、次の概念の例が含まれています。

• sparkSession の初期化
• データセットの読み取り
• データセットへの書き込み
• ローカルデータフレームの作成
• ExperienceEvent データのフィルター

SparkSession の初期化

すべての Scala ノートブックでは、次のテンプレートコードを使用してセッションを初期化する必要があります。

import org.apache.spark.sql.{ SparkSession }
val spark = SparkSession
  .builder()
  .master("local")
  .getOrCreate()

データセットの読み取り

Scala では、clientContext を読み込んで Platform の値を取得して返すことができるので、var userToken のような変数を定義する必要はありません。 以下の Scala の例では、clientContext を使用して、データセットの読み取りに必要なすべての値を取得し、返します。

import org.apache.spark.sql.{Dataset, SparkSession}
import com.adobe.platform.token.ClientContext
val spark = SparkSession.builder().master("local").config("spark.sql.warehouse.dir", "/").getOrCreate()

val clientContext = ClientContext.getClientContext()
val df1 = spark.read.format("com.adobe.platform.query")
  .option("user-token", clientContext.getUserToken())
  .option("ims-org", clientContext.getOrgId())
  .option("api-key", clientContext.getApiKey())
  .option("service-token", clientContext.getServiceToken())
  .option("sandbox-name", clientContext.getSandboxName())
  .option("mode", "batch")
  .option("dataset-id", "5e68141134492718af974844")
  .load()

df1.printSchema()
df1.show(10)

次の方法を使用して、JupyterLab buy で上記の例を自動生成できます。

JupyterLab の左側のナビゲーションで、「データアイコン」タブ（下でハイライト表示）を選択します。 Datasets ディレクトリと Schemas ディレクトリが表示されます。 「データセット」を選択して右クリックし、使用するデータセットのドロップダウンメニューから「ノートブック内のデータを調査」オプションを選択します。 ノートブックの下部に実行可能コードエントリが表示されます。
And

• 「Explore Data in Notebook」 を使用して、読み取りセルを生成します。
• 「Write Data in Notebook」を使用して、書き込みセルを生成します。

データセットへの書き込み

Scala では、clientContext を読み込んで Platform の値を取得して返すことができるので、var userToken のような変数を定義する必要はありません。 以下の Scala の例では、clientContext を使用して、データセットへの書き込みに必要なすべての値を定義し、返します。

import org.apache.spark.sql.{Dataset, SparkSession}
import com.adobe.platform.token.ClientContext
val spark = SparkSession.builder().master("local").config("spark.sql.warehouse.dir", "/").getOrCreate()

val clientContext = ClientContext.getClientContext()
df1.write.format("com.adobe.platform.query")
  .option("user-token", clientContext.getUserToken())
  .option("service-token", clientContext.getServiceToken())
  .option("ims-org", clientContext.getOrgId())
  .option("api-key", clientContext.getApiKey())
  .option("sandbox-name", clientContext.getSandboxName())
  .option("mode", "batch")
  .option("dataset-id", "5e68141134492718af974844")
  .save()

ローカルデータフレームの作成

Scala を使用してローカルデータフレームを作成するには、SQL クエリが必要です。 以下に例を示します。

sparkdf.createOrReplaceTempView("sparkdf")

val localdf = spark.sql("SELECT * FROM sparkdf LIMIT 1)

ExperienceEvent データをフィルタ

Scala ノートブック内の ExperienceEvent データセットにアクセスしてフィルタリングするには、データセット ID({DATASET_ID})、組織の IMS ID、および特定の時間範囲を定義するフィルタールールを指定する必要があります。 時間範囲のフィルタリングは、spark.sql() 関数を使用して定義します。関数パラメータは SQL クエリ文字列です。

次のセルでは、ExperienceEvent データセットを、2019 年 1 月 1 日から 2019 年 12 月 31 日の終わりまでの間にのみ存在するデータにフィルタリングします。

// Spark (Spark 2.4)

// Turn off extra logging
import org.apache.log4j.{Level, Logger}
Logger.getLogger("org").setLevel(Level.OFF)
Logger.getLogger("com").setLevel(Level.OFF)

import org.apache.spark.sql.{Dataset, SparkSession}
val spark = org.apache.spark.sql.SparkSession.builder().appName("Notebook")
  .master("local")
  .getOrCreate()

// Stage Exploratory
val dataSetId: String = "{DATASET_ID}"
val orgId: String = sys.env("IMS_ORG_ID")
val clientId: String = sys.env("PYDASDK_IMS_CLIENT_ID")
val userToken: String = sys.env("PYDASDK_IMS_USER_TOKEN")
val serviceToken: String = sys.env("PYDASDK_IMS_SERVICE_TOKEN")
val mode: String = "batch"

var df = spark.read.format("com.adobe.platform.query")
  .option("user-token", userToken)
  .option("ims-org", orgId)
  .option("api-key", clientId)
  .option("mode", mode)
  .option("dataset-id", dataSetId)
  .option("service-token", serviceToken)
  .load()
df.createOrReplaceTempView("event")
val timedf = spark.sql("""
    SELECT * 
    FROM event 
    WHERE timestamp > CAST('2019-01-01 00:00:00.0' AS TIMESTAMP)
    AND timestamp < CAST('2019-12-31 23:59:59.9' AS TIMESTAMP)
""")
timedf.show()

次の手順

このドキュメントでは、JupyterLab ノートブックを使用してデータセットにアクセスする際の一般的なガイドラインを説明しました。 データセットのクエリに関する詳細な例については、JupyterLab ノートブックの クエリサービス のドキュメントを参照してください。 データセットの調査と視覚化の方法について詳しくは、 ノートブック を使用したデータの分析に関するドキュメントを参照してください。

Query Service のオプションの SQL フラグ

次の表に、Query Service に使用できるオプションの SQL フラグの概要を示します。