azure - 如何在 Azure Synapse 中使用 PySpark 将文件装载为文件对象

Question

我有一个 azure 存储帐户(Storage gen2)，需要复制 config.yaml、文本文件、gz 文件等文件以在我的代码中引用它们。我已尝试 https://learn.microsoft.com/en-us/azure/synapse-analytics/spark/synapse-file-mount-api 中列出的步骤，但它的作用是挂载一个文件系统。如果您使用 for ex 引用它:yaml_file_test = mssparkutils.fs.head("synfs:/79/myMount/Test2/config.yaml",100)，它返回一个 Spark 数据帧而不是一个文件。

yaml 文件包含许多定义在整​​个项目中使用的局部变量。

我想要实现的是，如下所示。

with open('synfs:/80/myMount/Test2/config.yaml') as f:
    data = yaml.load(f, Loader=SafeLoader)
    print(data)

问题是 Pyspark 无法识别路径并给出错误:FileNotFoundError: [Errno 2] No such file or directory: 'synfs:/80/myMount/Test2/config.yaml'

我也必须以类似的方式访问其他文件，并将它们安装为文件对象来遍历并执行一些操作。例如，一些库(如 wordninja)需要“gz”文件而不是数据帧。当我尝试这样做时，出现上述错误。

如果我的方法不正确，任何人都可以帮助我们如何在 Azure Synapse 环境中实际创建全局变量以及如何从 Azure 存储实际创建文件对象。

请注意，我还尝试了其他从存储中读取数据的方法，如下所示，但问题是所有这些方法都在仅读取数据帧的路径中返回文件。

spark.conf.set("spark.storage.synapse.linkedServiceName", LinkService)
        spark.conf.set("fs.azure.account.oauth.provider.type", "com.microsoft.azure.synapse.tokenlibrary.LinkedServiceBasedTokenProvider")
        print("Connection Setup Successful!")
        return
    except Exception as e:
        print("Connection Setup Failed!- "+str(e))
        return -1

def spark_init(app_name: str = 'Mytest'):
    spark = SparkSession.builder.appName(app_name).getOrCreate()
    sc = spark.sparkContext
    return (spark, sc)

def getStream(streamsetlocation) :

  try:

    spark, sc = spark_init()
    setupConnection(spark,LinkService)
    print(streamsetlocation)
    dfStandardized = spark.read.format("csv").options(header=True).load(streamsetlocation)

任何帮助将不胜感激。

Answer 1

我无法获得上述挂载点来读取/写入二进制文件。但使用 fsspec 编写 Python pickle 文件并从 Azure Blob 存储中读回。

filename = 'final_model.sav'
sas_key = TokenLibrary.getConnectionString('')
storage_account_name = ‘’
container = ‘’
fsspec_handle = fsspec.open(f'abfs://{container}/{filename}', account_name = storage_account_name, sas_token=sas_key, mode='wb')
with fsspec_handle.open() as o_file:
pickle.dump(model, o_file)