Conexión manual a Flight service en los cuadernos de « Python »

Lectura de datos

Los pasos para leer datos de una fuente de datos son los siguientes:

1. Creación de un descriptor de vuelo con los metadatos necesarios para acceder a una fuente de datos
2. Creación de una instancia de un cliente de vuelos
3. Autenticación con Flight service
4. Envío del descriptor de vuelo a Flight service para obtener un objeto de información del vuelo
5. Lectura de datos de una fuente de datos. El fragmento de código muestra cómo leer en un pyarrow.Table, un pandas.DataFrame y cómo leer datos por partes.

Fragmento de código de ejemplo para leer datos. Si está trabajando con conjuntos de datos de gran tamaño, consulte las prácticas recomendadas para cargar grandes volúmenes de datos desde un archivo o una conexión.

# 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

Escribir datos

Los pasos para escribir datos en una fuente de datos son los siguientes:

1. Creación de un descriptor de vuelo con los metadatos necesarios para acceder a una fuente de datos
2. Creación de una instancia de un cliente de vuelos
3. Autenticación con Flight service
4. Obtener un flujo de escritura en vuelo
5. Escribir datos en un destino de datos

Fragmento de código de ejemplo para escribir datos:

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

Descripción del vuelo

Un elemento fundamental de la interacción con Flight service es el descriptor de vuelo, que especifica el acceso a la fuente de datos. Incluye una especificación de la fuente de datos en forma de propiedades de conexión, como el host, el puerto, etc., y propiedades de interacción, como el nombre de una tabla o una instrucción SQL. En lugar de las propiedades de conexión, también puedes especificar los ID del activo y del proyecto o del espacio de implementación.

Técnicamente, Flight service espera un descriptor de vuelo en forma de cadena JSON. En los cuadernos de « Python », puedes utilizar los diccionarios de « Python » para crear un descriptor de vuelo:

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)

Puede asset_id tratarse del identificador de un activo de datos conectado o de un activo de conexión.

Los descriptores de vuelo para leer o escribir datos solo difieren ligeramente en lo que respecta interaction_properties a... Las solicitudes de lectura suelen tener propiedades de interacción como «sql_statement», «file_name» o «table_name», mientras que las solicitudes de escritura tienen propiedades de interacción adicionales, como «existing_table_action» o «file_format».

Para obtener más información sobre la sintaxis de las solicitudes de datos, consulte «Solicitudes de datos de vuelo ».

Autenticación con el servidor de vuelos

Debes autenticarte en Flight service con un token de portador válido. Para ello, puedes utilizar el siguiente fragmento de código para escribir una clase de controlador de autenticación personalizada y crear una instancia de dicha clase.

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)