Manuelle Verbindung zu Flight service in „ Python “-Notebooks

Daten lesen

Die Schritte zum Auslesen von Daten aus einer Datenquelle umfassen:

1. Erstellen eines Flugdeskriptors mit den Metadaten für den Zugriff auf eine Datenquelle
2. Eine Instanz eines Flug-Clients erstellen
3. Authentifizierung über Flight service
4. Senden des Flugdeskriptors an Flight service, um ein Fluginformationsobjekt abzurufen
5. Daten aus einer Datenquelle lesen. Der Codeausschnitt zeigt, wie man in eine pyarrow.TableVariable liest pandas.DataFrame und wie man Daten in Blöcken liest.

Beispiel-Codeausschnitt zum Einlesen von Daten. Wenn Sie mit großen Datensätzen arbeiten, lesen Sie den Abschnitt „Bewährte Verfahren beim Laden großer Datenmengen aus einer Datei oder Verbindung “.

# 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

Daten schreiben

Die Schritte zum Schreiben von Daten in eine Datenquelle umfassen:

1. Erstellen eines Flugdeskriptors mit den Metadaten für den Zugriff auf eine Datenquelle
2. Eine Instanz eines Flug-Clients erstellen
3. Authentifizierung über Flight service
4. Einen Flug-Schreibstrom abrufen
5. Daten in ein Datenziel schreiben

Beispiel-Codeausschnitt zum Schreiben von Daten:

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()

Flugbeschreibung

Ein zentraler Bestandteil der Interaktion mit „ Flight service “ ist der Flugdeskriptor, der den Zugriff auf die Datenquelle festlegt. Sie enthält eine Datenquellenspezifikation in Form von Verbindungseigenschaften wie Host, Port usw. sowie Interaktionseigenschaften wie einen Tabellennamen oder eine SQL-Anweisung. Anstelle der Verbindungseigenschaften können Sie auch die IDs des Assets sowie des Projekts oder des Bereitstellungsbereichs angeben.

Genau genommen erwartet „ Flight service “ einen Flugdeskriptor in Form einer JSON-Zeichenkette. In „ Python “-Notebooks können Sie Wörterbücher aus „ Python “ verwenden, um einen Flugdeskriptor zu erstellen:

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)

Dies asset_id kann die ID eines verbundenen Datenobjekts oder eines Verbindungsobjekts sein.

Die Flugdeskriptoren für das Lesen oder Schreiben von Daten unterscheiden sich in dieser Hinsicht interaction_propertiesnur geringfügig. Leseanfragen verfügen in der Regel über Interaktionseigenschaften wie „sql_statement“, „file_name“ oder „table_name“, während Schreibanfragen zusätzliche Interaktionseigenschaften aufweisen, beispielsweise „existing_table_action“ oder „file_format“.

Weitere Informationen zur Syntax von Datenabfragen finden Sie unter „Flugdatenabfragen “.

Authentifizierung beim Flight Server

Sie müssen sich bei Flight service mit einem gültigen Bearer-Token authentifizieren. Zu diesem Zweck können Sie den folgenden Codeausschnitt verwenden, um eine benutzerdefinierte Authentifizierungs-Handler-Klasse zu schreiben und eine Instanz dieser Klasse zu erstellen.

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)