Python用Snowflakeコネクタを使用して結果をフェッチするワークロードの分散¶

概要¶

Cursor オブジェクトを使用してクエリを実行した後、結果バッチを使用して結果をフェッチする作業を分散できます。 結果バッチ は、結果のサブセットを取得する関数をカプセル化します。異なるワーカーを割り当てて異なる結果バッチを使用すると、結果を並行してフェッチおよび処理することができます。

結果バッチのリストの取得¶

クエリを実行した後、次のいずれかの形式で結果を取得できます。

• ResultBatch オブジェクト。

  これを実行するには、 カーソル オブジェクトで get_result_batches() メソッドを呼び出します。これにより、処理のためにさまざまなワーカーに割り当てることができる ResultBatch オブジェクトのリストが返されます。例:

  with connect(...) as conn: with conn.cursor() as cur: # Execute a query. cur.execute('select seq4() as n from table(generator(rowcount => 100000));') # Get the list of result batches result_batch_list = cur.get_result_batches() # Get the number of result batches in the list. num_result_batches = len(result_batch_list) # Split the list of result batches into two # to distribute the work of fetching results # between two workers. result_batch_list_1 = result_batch_list[:: 2] result_batch_list_2 = result_batch_list[1 :: 2] 

  Copy

• PyArrow テーブル。

  詳細については、 PyArrow テーブル をご参照ください。

  次のメソッドを使用して、結果バッチを PyArrow テーブルとして取得できます。

  • fetch_arrow_all(): このメソッドを呼び出して、すべての結果を含む PyArrow テーブルを返します。

  • fetch_arrow_batches(): このメソッドを呼び出して、各結果バッチの PyArrow テーブルを返すために使用できる反復子を返します。

  例:

  with connect(...) as conn: with conn.cursor() as cur: # Execute a query. cur.execute('select seq4() as n from table(generator(rowcount => 100000));') # Return a PyArrow table containing all of the results. table = cur.fetch_arrow_all() # Iterate over a list of PyArrow tables for result batches. for table_for_batch in cur.fetch_arrow_batches(): my_pyarrow_table_processing_function(table_for_batch) 

  Copy

• Pandas DataFrame オブジェクト。

  Python用SnowflakeコネクタのPandas互換バージョンをインストール している場合は、次の方法を使用して、結果バッチをPandas DataFrame オブジェクトとして取得できます。

  • fetch_pandas_all(): このメソッドを呼び出して、すべての結果を含むPandas DataFrame を返します。

  • fetch_pandas_batches(): このメソッドを呼び出して、各結果バッチのPandas DataFrame を返すために使用できる反復子を返します。

  例:

  with connect(...) as conn: with conn.cursor() as cur: # Execute a query. cur.execute('select seq4() as n from table(generator(rowcount => 100000));') # Return a pandas DataFrame containing all of the results. table = cur.fetch_pandas_all() # Iterate over a list of pandas DataFrames for result batches. for dataframe_for_batch in cur.fetch_pandas_batches(): my_dataframe_processing_function(dataframe_for_batch) 

  Copy

結果バッチのシリアル化¶

結果バッチを他のワーカーまたはノードに移動するために、結果バッチをシリアル化および逆シリアル化できます。例:

import pickle # Serialize a result batch from the first list. pickled_batch = pickle.dumps(result_batch_list_1[1]) # At this point, you can move the serialized data to # another worker/node. ... # Deserialize the result batch for processing. unpickled_batch = pickle.loads(pickled_batch) 

結果バッチの操作¶

次のセクションでは、 ResultBatch オブジェクトの操作方法について説明します。

• 結果バッチ内の行の反復処理

• 結果バッチの行の具体化

• 結果バッチの行数およびサイズの取得

• Arrow結果バッチの PyArrow テーブルまたはPandas DataFrame への変換

# Iterate over the list of result batches. for batch in result_batch_list_1: # Iterate over the subset of rows in a result batch. for row in batch: print(row) 

# Materialize the subset of results for the first result batch # in the list. first_result_batch = result_batch_list_1[1] first_result_batch_data = list(first_result_batch) 

# Get the number of rows in a result batch. num_rows = first_result_batch.rowcount # Get the size of the data in a result batch. compressed_size = first_result_batch.compressed_size uncompressed_size = first_result_batch.uncompressed_size 

with conn_cnx as con: with con.cursor() as cur: cur.execute("select col1 from table") batches = cur.get_result_batches() # Get the row from the ResultBatch as a pandas DataFrame. dataframe = batches[0].to_pandas() # Get the row from the ResultBatch as a PyArrow table. table = batches[0].to_arrow()