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

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

はじめに

このガイドを読む前に、JupyterLab ユーザーガイド ​ を参照して、JupyterLab の概要とデータサイエンスWorkspace内での役割を確認してください。

ノートブックのデータ制限 notebook-data-limits

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

Python および R の場合、ベンチマークには 40 GB RAM で設定されたノートブックサーバーが使用されました。 PySpark と Scala の場合、以下に概説するベンチマークには、64GB RAM、8 コア、2DBU、最大 4 ワーカーで構成されたデータブリッククラスターが使用されました。

使用される ExperienceEvent スキーマデータのサイズは、1,000 行（1K）から最大 10 億（1B）行まで様々でした。 PySpark および Spark 指標の場合、XDM データに 10 日の日付範囲が使用されました。

アドホックスキーマデータは、「テーブル Query Service 選択として作成」（CTAS）を使用して前処理されました。 また、このデータのサイズも、1000 行（1K）から最大 10 億行（1B）まで変化しました。

バッチモードとインタラクティブモードのどちらを使用するか mode

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

Python notebook のデータ制限

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

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

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

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

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

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

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

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

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

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

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

Python ノートブック python-notebook

Python ノートブックを使用すると、データセットにアクセスする際に、データにページ番号を付けることができます。 ページネーションを使用してデータを読み取るサンプルコードと使用しないサンプルデータを以下に示します。 使用可能なスターター Python ノートブックについて詳しくは、JupyterLab ユーザーガイドの JupyterLab ランチャー ​ の節を参照してください。

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

Python のデータセットからの読み取り python-read-dataset

ページネーションなし：

次のコードを実行すると、データセット全体が読み取られます。実行が成功した場合、データは 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 でのデータセットへの書き込み write-python

JupyterLab ノートブックのデータセットに書き込むには、JupyterLab の左側のナビゲーションにある「データ」アイコンタブ（以下でハイライト表示）を選択します。 Datasets ディレクトリと Schemas ディレクトリが表示されます。 「Datasets」を選択して右クリックし、使用するデータセットのドロップダウンメニューから「Write 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')

Query Service の Python を使用したクエリデータ query-data-python

JupyterLab 上の Experience Platform では、Python ノートブックで SQL を使用して、Adobe Experience Platform クエリサービス ​ 経由でデータにアクセスできます。 Query Service を通じたデータへのアクセスは実行時間が短いので、大規模なデータセットの処理に役立ちます。Query Service を使用したデータのクエリには 10 分間の処理時間制限があることに注意してください。

JupyterLab で Query Service を使用する前に、Query Service SQL 構文について実践的に理解していることを確認してください。

Query Service を使用したデータのクエリには、ターゲットデータセットの名前を指定する必要があります。 Data explorer を使用して目的のデータセットを見つけることで、必要なコードセルを生成できます。 データセットリストを右クリックし、「Query Data in Notebook」をクリックして、ノートブック内に 2 つのコードセルを生成します。 これら 2 つのセルについて、以下で詳しく説明します。

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

qs_connect()

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

%%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-filter

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

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

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

# 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-notebooks

r ノートブックを使用すると、データセットにアクセスする際にデータのページ番号を付けることができます。 ページネーションを使用してデータを読み取るサンプルコードと使用しないサンプルデータを以下に示します。 使用可能なスターター R ノートブックについて詳しくは、JupyterLab ユーザーガイドの JupyterLab ランチャー ​ の節を参照してください。

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

R のデータセットからの読み取り r-read-dataset

ページネーションなし：

次のコードを実行すると、データセット全体が読み取られます。実行が成功した場合、データは 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 のデータセットへの書き込み write-r

JupyterLab ノートブックのデータセットに書き込むには、JupyterLab の左側のナビゲーションにある「データ」アイコンタブ（以下でハイライト表示）を選択します。 Datasets ディレクトリと Schemas ディレクトリが表示されます。 「Datasets」を選択して右クリックし、使用するデータセットのドロップダウンメニューから「Write 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-filter

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

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

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

# 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-notebook

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

sparkSession の初期化 spark-initialize

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

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

%dataset を使用した PySpark 3 ノートブックの読み取りと書き込み magic

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

用途

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

説明

Data Science Workspace notebook （PySpark 3 カーネル）からデータセットを読み取ったり書き込んだりするためのカスタム Python magic コマンド。

例

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

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

ローカルデータフレームの作成 pyspark-create-dataframe

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-filter-experienceevent

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

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

# 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 ノートブック scala-notebook

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

SparkSession の初期化 scala-initialize

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

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

データセットの読み取り read-scala-dataset

Scala では、clientContext を読み込んでExperience 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 ディレクトリが表示されます。 「Datasets」を選択して右クリックし、使用するデータセットのドロップダウンメニューから「Explore Data in Notebook」オプションを選択します。 ノートブックの下部に実行可能コードエントリが表示されます。

および

データセットへの書き込み scala-write-dataset

Scala では、clientContext を読み込んでExperience 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-create-dataframe

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

sparkdf.createOrReplaceTempView("sparkdf")

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

ExperienceEvent データのフィルタリング scala-experienceevent

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

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

// 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 フラグ optional-sql-flags-for-query-service

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