Python ノートブックでの手動による Flight service への接続

オープン・ソースの pyarrow ライブラリーを使用して、Flight service を呼び出すことができます。プロジェクト内の データ資産 との間でデータの読み取りと書き込みを行うために独自のコードを作成する場合は、itc_utils ライブラリーを使用できます。

ノートブックの「コード・スニペット」ペインからファイルまたは接続からデータをロードするために生成されたコードを使用する場合、コードは pyarrow を使用して Flight service を呼び出します。また、オープン・ソースの pyarrow ライブラリーへの呼び出しをラップする itc_utils ライブラリーも使用します。これにより、追加のプログラミング作業を最小限に抑えながら、コードの読みやすさを向上させ、コード・サイズを削減します。

基本的な Flight service 対話

以下のコード・スニペットは、データ・ソースとの間でデータの読み取りと書き込みを行うための、 Flight service との基本的な対話を示しています。 このセクションには、 Flight Server での認証方法を示すコード・スニペットも含まれています。

データの読み取り

データ・ソースからデータを読み取るステップは、以下のとおりです。

  1. データ・ソースにアクセスするためのメタデータを使用したフライト記述子の作成
  2. フライト・クライアントのインスタンスの作成
  3. Flight service による認証
  4. Flight service へのフライト記述子の送信によるフライト情報オブジェクトの取得
  5. データ・ソースからのデータの読み取り。 コード・スニペットは、 pyarrow.Table 、 pandas.DataFrame への読み取り、およびデータのチャンク単位での読み取り方法を示しています。

データを読み取るための説明用のサンプル・コード・スニペット。 大きなデータ・セットを処理する場合は、 ファイルまたは接続から大量のデータをロードする際のベスト・プラクティス を参照してください。

# create a flight descriptor specifiying the data source or target
# content and structure of cmd are specific to IBM CP4D's flight service.
flightDescriptor = pyarrow.flight.FlightDescriptor.for_command(cmd)

# create an instance of a flight client
flightClient = pyarrow.flight.FlightClient(url, **opts)

# authenticate with the flight service (for authHandler, see code snippet at the end of this section).
flightClient.authenticate(authHandler)

# send the flight descriptor the flight service to obtain a FlightInfo object
# which provides information for reading data from one or more endpoints.
flightInfo = flightClient.get_flight_info(flightDescriptor)

# read from all endpoints
for endpoint in flightInfo.endpoints:
    reader = flightClient.do_get(endpoint.ticket)

# from an endpoint (or reader, or stream), you can read in several ways:
# 1) read a pyarrow.Table
table = reader.read_all()
# 2) read a pandas.DataFrame
df = reader.read_pandas()
# 3) read in chunks, i.e. a number of pyarrow.flight.RecordBatch
while True:
    try:
        recordBatch = reader.read_chunk() # read a pyarrow.flight.RecordBatch
    except StopIteration:
        break

データの書き込み

データ・ソースにデータを書き込むステップは、以下のとおりです。

  1. データ・ソースにアクセスするためのメタデータを使用したフライト記述子の作成
  2. フライト・クライアントのインスタンスの作成
  3. Flight service による認証
  4. フライト書き込みストリームの取得
  5. データ・ターゲットへのデータの書き込み

データを書き込むためのサンプル・コード・スニペット:

import pyarrow as pa

# create a flight descriptor specifiying the data source or target
# content and structure of cmd are specific to IBM CP4D's flight service.
flightDescriptor = pyarrow.flight.FlightDescriptor.for_command(cmd)

# create an instance of a flight client
flightClient = pyarrow.flight.FlightClient(url, **opts)

# authenticate with Flight service (for authHandler, see code snippet at the end of this section).
flightClient.authenticate(authHandler)

# obtain a flight write stream
schema = pa.Schema.from_pandas(df, preserve_index=False)
writer, reader = flightClient.do_put(flightDescriptor, schema)

# write data to a data target
writer.write_table(pa.Table.from_pandas(df, schema))
writer.close()

フライト記述子

Flight service との対話の中心的な部分は、データ・ソースへのアクセスを指定するフライト記述子です。 これには、接続プロパティー (ホスト、ポートなど) および対話プロパティー (表名や SQL ステートメント など) の形式でデータ・ソース仕様が含まれます。 接続プロパティーの代わりに、 資産 の ID と、プロジェクトまたは デプロイメント・スペース の ID を指定することもできます。

技術的には、 Flight service は JSON ストリングの形式のフライト記述子を予期します。 Python ・ノートブックでは、 Python 辞書を使用して、以下のようにフライト記述子を構成できます。

flight_request = {
    "asset_id": "<asset_id>",
    "project_id": "<project_id>",
    "interaction_properties": {
        "schema_name": "<schema>",
        "table_name": "<table>",
        "row_limit": 5000
    }
}

# create a flight descriptor
cmd = json.dumps(flight_request)
flightDescriptor = pyarrow.flight.FlightDescriptor.for_command(cmd)

asset_id は、接続済みの データ資産 または接続 資産 の ID にすることができます。

データの読み取りまたは書き込みのためのフライト・ディスクリプターは、 interaction_properties に関してはわずかに異なります。 読み取り要求には通常、「 sql_ステートメント 」、「 file_name 」、または「 table_name 」などの相互作用プロパティーが含まれますが、書き込み要求には「 existing_table_action 」や「 file_format 」などの追加の相互作用プロパティーが含まれます。

データ要求の構文について詳しくは、 フライト・データ要求 を参照してください。

Flight Server での認証

有効なベアラー・トークンを使用して Flight service で認証する必要があります。 そのために、以下のコード・スニペットを使用して、カスタム認証ハンドラー・クラスを作成し、このクラスのインスタンスを作成することができます。

import pyarrow.flight as flight

class TokenClientAuthHandler(flight.ClientAuthHandler):
    """An example implementation of authentication with a user token."""

    def __init__(self, token):
        super().__init__()
        strToken = str(token)
        self.token = strToken.encode('utf-8')

    def authenticate(self, outgoing, incoming):
        outgoing.write(self.token)
        self.token = incoming.read()

    def get_token(self):
        return self.token

# create an instance of the authentication handler by using IBM Watson Studio Lib

from ibm_watson_studio_lib import access_project_or_space
wslib = access_project_or_space()
token = 'Bearer {}'.format(wslib.auth.get_current_token())

authHandler = TokenClientAuthHandler(token)

詳細情報