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

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

概要

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

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

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

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

ExperienceEventスキーマデータのサイズは、1,000行（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スキーマ： 最大100万行のXDMデータ（ディスク上の3 GBデータ）を13分以内に読み取ることができます。

アドホックスキーマ： アドホックデータの最大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フラグの概要を示します。