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

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

はじめに

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

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

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

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

使用されるExperienceEvent スキーマデータのサイズは、100 （1K）行から10億（1B）行まで様々です。 PySparkおよびSpark指標の場合、XDM データには10日間の日付範囲が使用されていることに注意してください。

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

バッチモードとインタラクティブモードの使用例 mode

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

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

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

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

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

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

アドホックスキーマ：​最大300万行のアドホックデータ（ディスク上の293MB データ）を約10分で読み取ることができます。

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

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

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

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

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

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

Python ノートブック python-notebook

Python ノートブックを使用すると、データセットにアクセスする際にデータをページネーションできます。 ページネーションの有無にかかわらずデータを読み取るサンプルコードを以下に示します。 利用可能なスターターPython ノートブックについて詳しくは、JupyterLab ユーザーガイド内のJupyterLab Launcher セクションを参照してください。

以下の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 を通じたデータへのアクセスは実行時間が短いので、大規模なデータセットの処理に役立ちます。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 データセットにアクセスしてフィルタリングするには、データセットのID （{DATASET_ID}）と、論理演算子を使用して特定の時間範囲を定義するフィルタールールを指定する必要があります。 時間範囲を定義すると、指定されたページ番号は無視され、データセット全体が考慮されます。

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

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

# 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 ノートブックを使用すると、データセットにアクセスする際にデータをページネーションできます。 ページネーションの有無にかかわらずデータを読み取るサンプルコードを以下に示します。 利用可能なスターターまたはノートブックについて詳しくは、JupyterLab ユーザーガイド内のJupyterLab Launcher セクションを参照してください。

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

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

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

# 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 ノートブック （PySpark 3 カーネル）からデータセットを読み取るか書き込むためのカスタム Python マジックコマンド。

例

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

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 クエリ文字列です。

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

# 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 クエリ文字列です。

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

// 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 フラグの概要を示しています。