ダウンストリームRAGの使用例

備考

プレビュー

Microsoft SharePoint コネクタはベータ版です。

SharePoint パイプラインを作成したので、生のドキュメントをテキストに解析したり、解析されたデータをチャンク化したり、チャンクから埋め込みを作成したりできます。その後、出力テーブルで readStream をダウンストリームパイプラインで直接使用できます。

ファイルデータにアクセスするために、次のファイルアクセス UDF を提供しています。これらの UDF は、インジェストパイプラインからの出力テーブルで実行できます。

名前	説明
`read_blob_as_file`(blob カラム、ファイル名カラム)	ファイルをローカルディスクにダウンロードし、ファイルパスを返します。
`read_blob_as_bytes`(ブロブ列)	ファイルをローカルディスクにダウンロードし、データをバイト配列として返します。

ファイルアクセス UDF の設定

次のセルをダウンストリームパイプラインに追加します。

Python
# DO NOT MODIFY this cell.

from pyspark.sql.functions import udf, struct
from pyspark.sql.types import BinaryType

# Copy to local disk and get file path.

def copy_to_disk(blob, filename) -> str:
fname = "/local_disk0/tmp/" + filename
with open(fname, "wb") as f:
f.write(blob.inline_content)
return fname

read_blob_as_file = udf(copy_to_disk)

# Get bytes directly.

def get_bytes(blob) -> bytes:
return blob.inline_content

read_blob_as_bytes = udf(get_bytes, BinaryType())

例

ファイルパスを返すには:

Python
# Suppose you have a simple UDF that converts a file's raw bytes to a UTF-8 string.

def file_bytes_to_text(fname):
with open(fname, "rb") as f:
return f.read().decode("utf-8")
file_bytes_to_text_udf = udf(file_bytes_to_text)

# Chain your UDF with the file access UDF for the file path.

df.withColumn("text_content", file_bytes_to_text_udf(read_blob_as_file("content", "file_metadata.name"))).collect()

データをバイト配列として返すには、次のようにします。

Python
# Suppose you have a simple UDF that converts a file's raw bytes to a UTF-8 string.

def bytes_to_text(bytes_data):
return bytes_data.decode("utf-8")
bytes_to_text_udf = udf(bytes_to_text)

# Chain your UDF with the file access UDF for the byte array.

df.withColumn("text_content", bytes_to_text_udf(read_blob_as_bytes("content"))).collect()

注記

ファイルアクセス UDF は、100 MB を超えるファイルのファイルコンテンツを処理できません。ファイルアクセス UDF を使用する前に、これらの行を除外する必要があります。

ファイル・パスはローカル・ディスクに書き込まれ UDF ため、シングル・ユーザー・クラスターでのみ機能します。代わりにクラシッククラスターまたはサーバレスコンピュートでダウンストリームパイプラインを実行する場合は、ローカルディスクではなくUnity Catalogボリュームに書き込むようにUDFを更新できます。ただし、これによりパフォーマンスが低下します。

ボリュームに書き込むには、次のようにします。

Python
# Update the volume_path in the function below.
from pyspark.sql.functions import udf, struct


# copy to volume_path and get file path
def copy_to_disk(blob, filename) -> str:
  # UPDATE THIS VALUE
  volume_path = "/Volumes/<my_catalog>/<my schema>/<my volume name>/"


  fname = volume_path + filename
  with open(fname, "wb") as f:
    f.write(blob.inline_content)
  return fname


read_blob_as_file = udf(copy_to_disk)

ファイルアクセス UDF の設定​

例​

ファイルアクセス UDF の設定

例