Microsoft Spark 实用工具简介

from pyspark.sql import SparkSession

# Azure storage access info
blob_account_name = 'Your account name' # replace with your blob name
blob_container_name = 'Your container name' # replace with your container name
blob_relative_path = 'Your path' # replace with your relative folder path
linked_service_name = 'Your linked service name' # replace with your linked service name

blob_sas_token = mssparkutils.credentials.getConnectionStringOrCreds(linked_service_name)

# Allow SPARK to access from Blob remotely

wasb_path = 'wasbs://%s@%s.blob.core.chinacloudapi.cn/%s' % (blob_container_name, blob_account_name, blob_relative_path)

spark.conf.set('fs.azure.sas.%s.%s.blob.core.chinacloudapi.cn' % (blob_container_name, blob_account_name), blob_sas_token)
print('Remote blob path: ' + wasb_path)

val blob_account_name = "" // replace with your blob name
val blob_container_name = "" //replace with your container name
val blob_relative_path = "/" //replace with your relative folder path
val linked_service_name = "" //replace with your linked service name


val blob_sas_token = mssparkutils.credentials.getConnectionStringOrCreds(linked_service_name)

val wasbs_path = f"wasbs://$blob_container_name@$blob_account_name.blob.core.chinacloudapi.cn/$blob_relative_path"
spark.conf.set(f"fs.azure.sas.$blob_container_name.$blob_account_name.blob.core.chinacloudapi.cn",blob_sas_token)

var blob_account_name = "";  // replace with your blob name
var blob_container_name = "";     // replace with your container name
var blob_relative_path = "";  // replace with your relative folder path
var linked_service_name = "";    // replace with your linked service name
var blob_sas_token = Credentials.GetConnectionStringOrCreds(linked_service_name);

spark.Conf().Set($"fs.azure.sas.{blob_container_name}.{blob_account_name}.blob.core.chinacloudapi.cn", blob_sas_token);

var wasbs_path = $"wasbs://{blob_container_name}@{blob_account_name}.blob.core.chinacloudapi.cn/{blob_relative_path}";

Console.WriteLine(wasbs_path);

# Azure storage access info
blob_account_name <- 'Your account name' # replace with your blob name
blob_container_name <- 'Your container name' # replace with your container name
blob_relative_path <- 'Your path' # replace with your relative folder path
linked_service_name <- 'Your linked service name' # replace with your linked service name

blob_sas_token <- mssparkutils.credentials.getConnectionStringOrCreds(linked_service_name)

# Allow SPARK to access from Blob remotely
sparkR.session()
wasb_path <- sprintf('wasbs://%s@%s.blob.core.chinacloudapi.cn/%s',blob_container_name, blob_account_name, blob_relative_path)
sparkR.session(sprintf('fs.azure.sas.%s.%s.blob.core.chinacloudapi.cn',blob_container_name, blob_account_name), blob_sas_token)

print( paste('Remote blob path: ',wasb_path))

from notebookutils import mssparkutils
mssparkutils.fs.help()

mssparkutils.fs.help()

using Microsoft.Spark.Extensions.Azure.Synapse.Analytics.Notebook.MSSparkUtils;
FS.Help()

library(notebookutils)
mssparkutils.fs.help()

mssparkutils.fs.ls('Your directory path')

mssparkutils.fs.ls("Your directory path")

FS.Ls("Your directory path")

mssparkutils.fs.ls("Your directory path")

files = mssparkutils.fs.ls('Your directory path')
for file in files:
    print(file.name, file.isDir, file.isFile, file.path, file.size, file.modifyTime)

val files = mssparkutils.fs.ls("/")
files.foreach{
    file => println(file.name,file.isDir,file.isFile,file.size,file.modifyTime)
}

var Files = FS.Ls("/");
foreach(var File in Files) {
    Console.WriteLine(File.Name+" "+File.IsDir+" "+File.IsFile+" "+File.Size);
}

files <- mssparkutils.fs.ls("/")
for (file in files) {
    writeLines(paste(file$name, file$isDir, file$isFile, file$size, file$modifyTime))
}

mssparkutils.fs.mkdirs('new directory name')

mssparkutils.fs.mkdirs("new directory name")

FS.Mkdirs("new directory name")

mssparkutils.fs.mkdirs("new directory name")

mssparkutils.fs.cp('source file or directory', 'destination file or directory', True)# Set the third parameter as True to copy all files and directories recursively

mssparkutils.fs.cp("source file or directory", "destination file or directory", true) // Set the third parameter as True to copy all files and directories recursively

FS.Cp("source file or directory", "destination file or directory", true) // Set the third parameter as True to copy all files and directories recursively

mssparkutils.fs.cp('source file or directory', 'destination file or directory', True)

mssparkutils.fs.head('file path', maxBytes to read)

mssparkutils.fs.head("file path", maxBytes to read)

FS.Head("file path", maxBytes to read)

mssparkutils.fs.head('file path', maxBytes to read)

mssparkutils.fs.mv('source file or directory', 'destination directory', True) # Set the last parameter as True to firstly create the parent directory if it does not exist

mssparkutils.fs.mv("source file or directory", "destination directory", true) // Set the last parameter as True to firstly create the parent directory if it does not exist

FS.Mv("source file or directory", "destination directory", true)

mssparkutils.fs.mv('source file or directory', 'destination directory', True) # Set the last parameter as True to firstly create the parent directory if it does not exist

mssparkutils.fs.put("file path", "content to write", True) # Set the last parameter as True to overwrite the file if it existed already

mssparkutils.fs.put("file path", "content to write", true) // Set the last parameter as True to overwrite the file if it existed already

FS.Put("file path", "content to write", true) // Set the last parameter as True to overwrite the file if it existed already

mssparkutils.fs.put("file path", "content to write", True) # Set the last parameter as True to overwrite the file if it existed already

mssparkutils.fs.append("file path", "content to append", True) # Set the last parameter as True to create the file if it does not exist

mssparkutils.fs.append("file path","content to append",true) // Set the last parameter as True to create the file if it does not exist

FS.Append("file path", "content to append", true) // Set the last parameter as True to create the file if it does not exist

mssparkutils.fs.append("file path", "content to append", True) # Set the last parameter as True to create the file if it does not exist

mssparkutils.fs.rm('file path', True) # Set the last parameter as True to remove all files and directories recursively

mssparkutils.fs.rm("file path", true) // Set the last parameter as True to remove all files and directories recursively

FS.Rm("file path", true) // Set the last parameter as True to remove all files and directories recursively

mssparkutils.fs.rm('file path', True) # Set the last parameter as True to remove all files and directories recursively

不支持。

可以使用 MSSparkUtils 笔记本实用工具运行笔记本或使用值退出笔记本。 运行以下命令以概要了解可用的方法：

mssparkutils.notebook.help()

获取结果：

The notebook module.

exit(value: String): void -> This method lets you exit a notebook with a value.
run(path: String, timeoutSeconds: int, arguments: Map): String -> This method runs a notebook and returns its exit value.

引用笔记本

引用笔记本并返回其退出值。 可以在笔记本中以交互方式或在管道中运行嵌套函数调用。 所引用的笔记本将在其调用此函数的 Spark 池上运行。

mssparkutils.notebook.run("notebook path", <timeoutSeconds>, <parameterMap>)

例如：

mssparkutils.notebook.run("folder/Sample1", 90, {"input": 20 })

运行完成后，你将看到名为“查看笔记本运行：笔记本名称”的快照链接，显示在单元格输出中，可以单击该链接以查看此特定运行的快照。

并行引用运行多个笔记本

mssparkutils.notebook.runMultiple() 方法让你可以并行运行多个笔记本，或使用预定义的拓扑结构。 该 API 在 Spark 会话中使用多线程实现机制，这意味着计算资源由引用笔记本运行共享。

通过 mssparkutils.notebook.runMultiple()，您可以：

• 同时执行多个笔记本，而无需等待每个笔记本完成。

• 使用简单的 JSON 格式为笔记本指定依赖项和执行顺序。

• 优化 Spark 计算资源的使用并降低 Synapse 项目的成本。

• 在输出中查看每个笔记本运行记录的快照，并方便地调试/监视笔记本任务。

• 获取每个执行活动的退出值，并在下游任务中使用它们。

还可以尝试运行 mssparkutils.notebook.help("runMultiple") 来查找示例和详细用法。

下面是使用此方法并行运行一组笔记本的简单示例：

mssparkutils.notebook.runMultiple(["NotebookSimple", "NotebookSimple2"])

根笔记本中的执行结果如下所示：

下面是使用 mssparkutils.notebook.runMultiple() 运行具有拓扑结构的笔记本的示例。 使用此方法通过代码体验轻松编排笔记本。

# run multiple notebooks with parameters
DAG = {
    "activities": [
        {
            "name": "NotebookSimple", # activity name, must be unique
            "path": "NotebookSimple", # notebook path
            "timeoutPerCellInSeconds": 90, # max timeout for each cell, default to 90 seconds
            "args": {"p1": "changed value", "p2": 100}, # notebook parameters
        },
        {
            "name": "NotebookSimple2",
            "path": "NotebookSimple2",
            "timeoutPerCellInSeconds": 120,
            "args": {"p1": "changed value 2", "p2": 200}
        },
        {
            "name": "NotebookSimple2.2",
            "path": "NotebookSimple2",
            "timeoutPerCellInSeconds": 120,
            "args": {"p1": "changed value 3", "p2": 300},
            "retry": 1,
            "retryIntervalInSeconds": 10,
            "dependencies": ["NotebookSimple"] # list of activity names that this activity depends on
        }
    ]
}
mssparkutils.notebook.runMultiple(DAG)

退出笔记本

使用值退出笔记本。 可以在笔记本中以交互方式或在管道中运行嵌套函数调用。

• 在笔记本中以交互方式调用 exit() 函数时，Azure Synapse 将引发异常、跳过运行子序列单元格并使 Spark 会话保持活动状态。

• 协调在 Synapse 管道中调用 exit() 函数的笔记本时，Azure Synapse 将返回退出值、完成管道运行并停止 Spark 会话。

• 在所引用的笔记本中调用 exit() 函数时，Azure Synapse 将在其中停止进一步的执行，并继续运行调用 run() 函数的笔记本中的下一个单元格。 例如：Notebook1 有三个单元格，调用第二个单元格中的 exit() 函数。 Notebook2 有五个单元格，调用第三个单元格中的 run(notebook1) 函数。 运行 Notebook2 时，如果命中 exit() 函数，Notebook1 将在第二个单元格停止。 Notebook2 将继续运行其第四和第五个单元格。

mssparkutils.notebook.exit("value string")

例如：

Sample1 笔记本在 folder/ 下查找以下两个单元格 ：

• 单元格 1 定义 input 参数，默认值设为 10。
• 单元格 2 退出笔记本，input 作为退出值。

可以使用默认值在另一笔记本中运行 Sample1：

exitVal = mssparkutils.notebook.run("folder/Sample1")
print (exitVal)

结果：

Sample1 run success with input is 10

可以在另一笔记本中运行 Sample1，并将 input 值设为 20 ：

exitVal = mssparkutils.notebook.run("mssparkutils/folder/Sample1", 90, {"input": 20 })
print (exitVal)

结果：

Sample1 run success with input is 20

可以使用 MSSparkUtils 笔记本实用工具运行笔记本或使用值退出笔记本。 运行以下命令以概要了解可用的方法：

mssparkutils.notebook.help()

获取结果：

The notebook module.

exit(value: String): void -> This method lets you exit a notebook with a value.
run(path: String, timeoutSeconds: int, arguments: Map): String -> This method runs a notebook and returns its exit value.

引用笔记本

引用笔记本并返回其退出值。 可以在笔记本中以交互方式或在管道中运行嵌套函数调用。 所引用的笔记本将在其调用此函数的 Spark 池上运行。

mssparkutils.notebook.run("notebook path", <timeoutSeconds>, <parameterMap>)

例如：

mssparkutils.notebook.run("folder/Sample1", 90, Map("input" -> 20))

运行完成后，你将看到名为“查看笔记本运行：笔记本名称”的快照链接，显示在单元格输出中，可以单击该链接以查看此特定运行的快照。

退出笔记本

使用值退出笔记本。 可以在笔记本中以交互方式或在管道中运行嵌套函数调用。

• 在笔记本中以交互方式调用 exit() 函数时，Azure Synapse 将引发异常、跳过运行子序列单元格并使 Spark 会话保持活动状态。

• 协调在 Synapse 管道中调用 exit() 函数的笔记本时，Azure Synapse 将返回退出值、完成管道运行并停止 Spark 会话。

• 在所引用的笔记本中调用 exit() 函数时，Azure Synapse 将在其中停止进一步的执行，并继续运行调用 run() 函数的笔记本中的下一个单元格。 例如：Notebook1 有三个单元格，调用第二个单元格中的 exit() 函数。 Notebook2 有五个单元格，调用第三个单元格中的 run(notebook1) 函数。 运行 Notebook2 时，如果命中 exit() 函数，Notebook1 将在第二个单元格停止。 Notebook2 将继续运行其第四和第五个单元格。

mssparkutils.notebook.exit("value string")

例如：

Sample1 笔记本在 mssparkutils/folder/ 下查找以下两个单元格 ：

• 单元格 1 定义 input 参数，默认值设为 10。
• 单元格 2 退出笔记本，input 作为退出值。

可以使用默认值在另一笔记本中运行 Sample1：

val exitVal = mssparkutils.notebook.run("mssparkutils/folder/Sample1")
print(exitVal)

结果：

exitVal: String = Sample1 run success with input is 10
Sample1 run success with input is 10

可以在另一笔记本中运行 Sample1，并将 input 值设为 20 ：

val exitVal = mssparkutils.notebook.run("mssparkutils/folder/Sample1", 90, {"input": 20 })
print(exitVal)

结果：

exitVal: String = Sample1 run success with input is 20
Sample1 run success with input is 20

可以使用 MSSparkUtils 笔记本实用工具运行笔记本或使用值退出笔记本。 运行以下命令以概要了解可用的方法：

mssparkutils.notebook.help()

获取结果：

The notebook module.

exit(value: String): void -> This method lets you exit a notebook with a value.
run(path: String, timeoutSeconds: int, arguments: Map): String -> This method runs a notebook and returns its exit value.

引用笔记本

引用笔记本并返回其退出值。 可以在笔记本中以交互方式或在管道中运行嵌套函数调用。 所引用的笔记本将在其调用此函数的 Spark 池上运行。

mssparkutils.notebook.run("notebook path", <timeoutSeconds>, <parameterMap>)

例如：

mssparkutils.notebook.run("folder/Sample1", 90, list("input": 20))

运行完成后，你将看到名为“查看笔记本运行：笔记本名称”的快照链接，显示在单元格输出中，可以单击该链接以查看此特定运行的快照。

退出笔记本

使用值退出笔记本。 可以在笔记本中以交互方式或在管道中运行嵌套函数调用。

• 在笔记本中以交互方式调用 exit() 函数时，Azure Synapse 将引发异常、跳过运行子序列单元格并使 Spark 会话保持活动状态。

• 协调在 Synapse 管道中调用 exit() 函数的笔记本时，Azure Synapse 将返回退出值、完成管道运行并停止 Spark 会话。

• 在所引用的笔记本中调用 exit() 函数时，Azure Synapse 将在其中停止进一步的执行，并继续运行调用 run() 函数的笔记本中的下一个单元格。 例如：Notebook1 有三个单元格，调用第二个单元格中的 exit() 函数。 Notebook2 有五个单元格，调用第三个单元格中的 run(notebook1) 函数。 运行 Notebook2 时，如果命中 exit() 函数，Notebook1 将在第二个单元格停止。 Notebook2 将继续运行其第四和第五个单元格。

mssparkutils.notebook.exit("value string")

例如：

Sample1 笔记本在 folder/ 下查找以下两个单元格 ：

• 单元格 1 定义 input 参数，默认值设为 10。
• 单元格 2 退出笔记本，input 作为退出值。

可以使用默认值在另一笔记本中运行 Sample1：

exitVal <- mssparkutils.notebook.run("folder/Sample1")
print (exitVal)

结果：

Sample1 run success with input is 10

可以在另一笔记本中运行 Sample1，并将 input 值设为 20 ：

exitVal <- mssparkutils.notebook.run("mssparkutils/folder/Sample1", 90, list("input": 20))
print (exitVal)

结果：

Sample1 run success with input is 20

mssparkutils.credentials.help()

mssparkutils.credentials.help()

Not supported.

mssparkutils.credentials.help()

getToken(audience, name): returns AAD token for a given audience, name (optional)
isValidToken(token): returns true if token hasn't expired
getConnectionStringOrCreds(linkedService): returns connection string or credentials for linked service
getFullConnectionString(linkedService): returns full connection string with credentials
getPropertiesAll(linkedService): returns all the properties of a linked servicegetSecret(akvName, secret, linkedService): returns AKV secret for a given AKV linked service, akvName, secret key
getSecret(akvName, secret): returns AKV secret for a given akvName, secret key
getSecretWithLS(linkedService, secret): returns AKV secret for a given linked service, secret key
putSecret(akvName, secretName, secretValue, linkedService): puts AKV secret for a given akvName, secretName
putSecret(akvName, secretName, secretValue): puts AKV secret for a given akvName, secretName
putSecretWithLS(linkedService, secretName, secretValue): puts AKV secret for a given linked service, secretName

getToken(audience, name): returns AAD token for a given audience, name (optional)
isValidToken(token): returns true if token hasn't expired
getConnectionStringOrCreds(linkedService): returns connection string or credentials for linked service
getFullConnectionString(linkedService): returns full connection string with credentials
getPropertiesAll(linkedService): returns all the properties of a linked servicegetSecret(akvName, secret, linkedService): returns AKV secret for a given AKV linked service, akvName, secret key
getSecret(akvName, secret): returns AKV secret for a given akvName, secret key
getSecretWithLS(linkedService, secret): returns AKV secret for a given linked service, secret key
putSecret(akvName, secretName, secretValue, linkedService): puts AKV secret for a given akvName, secretName
putSecret(akvName, secretName, secretValue): puts AKV secret for a given akvName, secretName
putSecretWithLS(linkedService, secretName, secretValue): puts AKV secret for a given linked service, secretName

getToken(audience, name): returns AAD token for a given audience, name (optional)
isValidToken(token): returns true if token hasn't expired
getConnectionStringOrCreds(linkedService): returns connection string or credentials for linked service
getFullConnectionString(linkedService): returns full connection string with credentials
getPropertiesAll(linkedService): returns all the properties of a linked servicegetSecret(akvName, secret, linkedService): returns AKV secret for a given AKV linked service, akvName, secret key
getSecret(akvName, secret): returns AKV secret for a given akvName, secret key
putSecret(akvName, secretName, secretValue, linkedService): puts AKV secret for a given akvName, secretName
putSecret(akvName, secretName, secretValue): puts AKV secret for a given akvName, secretName
putSecretWithLS(linkedService, secretName, secretValue): puts AKV secret for a given linked service, secretName

getToken(audience, name): returns AAD token for a given audience, name (optional)
isValidToken(token): returns true if token hasn't expired
getConnectionStringOrCreds(linkedService): returns connection string or credentials for linked service
getFullConnectionString(linkedService): returns full connection string with credentials
getPropertiesAll(linkedService): returns all the properties of a linked servicegetSecret(akvName, secret, linkedService): returns AKV secret for a given AKV linked service, akvName, secret key
getSecret(akvName, secret): returns AKV secret for a given akvName, secret key
getSecretWithLS(linkedService, secret): returns AKV secret for a given linked service, secret key
putSecret(akvName, secretName, secretValue, linkedService): puts AKV secret for a given akvName, secretName
putSecret(akvName, secretName, secretValue): puts AKV secret for a given akvName, secretName
putSecretWithLS(linkedService, secretName, secretValue): puts AKV secret for a given linked service, secretName

mssparkutils.credentials.getToken('audience Key')

mssparkutils.credentials.getToken("audience Key")

Credentials.GetToken("audience Key")

mssparkutils.credentials.getToken('audience Key')

mssparkutils.credentials.isValidToken('your token')

mssparkutils.credentials.isValidToken("your token")

Credentials.IsValidToken("your token")

mssparkutils.credentials.isValidToken('your token')

mssparkutils.credentials.getConnectionStringOrCreds('linked service name')

mssparkutils.credentials.getConnectionStringOrCreds("linked service name")

Credentials.GetConnectionStringOrCreds("linked service name")

mssparkutils.credentials.getConnectionStringOrCreds('linked service name')

mssparkutils.credentials.getSecret('azure key vault name','secret name','linked service name')

mssparkutils.credentials.getSecret("azure key vault name","secret name","linked service name")

Credentials.GetSecret("azure key vault name","secret name","linked service name")

mssparkutils.credentials.getSecret('azure key vault name','secret name','linked service name')

mssparkutils.credentials.getSecret('azure key vault name','secret name')

mssparkutils.credentials.getSecret("azure key vault name","secret name")

Credentials.GetSecret("azure key vault name","secret name")

mssparkutils.credentials.getSecret('azure key vault name','secret name')

使用工作区标识放置机密

使用工作区标识放置给定 Azure Key Vault 名称、机密名称和链接服务名称的 Azure Key Vault 机密。 请确保正确配置对 Azure Key Vault 的访问。

mssparkutils.credentials.putSecret('azure key vault name','secret name','secret value','linked service name')

使用工作区标识放置机密

使用工作区标识放置给定 Azure Key Vault 名称、机密名称和链接服务名称的 Azure Key Vault 机密。 请确保正确配置对 Azure Key Vault 的访问。

mssparkutils.credentials.putSecret("azure key vault name","secret name","secret value","linked service name")

使用工作区标识放置机密

使用工作区标识放置给定 Azure Key Vault 名称、机密名称和链接服务名称的 Azure Key Vault 机密。 请确保正确配置对 Azure Key Vault 的访问。

mssparkutils.credentials.putSecret('azure key vault name','secret name','secret value','linked service name')

使用用户凭据放置机密

使用用户凭据放置给定 Azure Key Vault 名称、机密名称和链接服务名称的 Azure Key Vault 机密。

mssparkutils.credentials.putSecret('azure key vault name','secret name','secret value')

使用用户凭据放置机密

使用用户凭据放置给定 Azure Key Vault 名称、机密名称和链接服务名称的 Azure Key Vault 机密。

mssparkutils.credentials.putSecret('azure key vault name','secret name','secret value')

使用用户凭据放置机密

使用用户凭据放置给定 Azure Key Vault 名称、机密名称和链接服务名称的 Azure Key Vault 机密。

mssparkutils.credentials.putSecret("azure key vault name","secret name","secret value")

mssparkutils.env.help()

mssparkutils.env.help()

mssparkutils.env.help()

Env.Help()

mssparkutils.env.getUserName()

mssparkutils.env.getUserName()

mssparkutils.env.getUserName()

Env.GetUserName()

mssparkutils.env.getUserId()

mssparkutils.env.getUserId()

mssparkutils.env.getUserId()

Env.GetUserId()

mssparkutils.env.getJobId()

mssparkutils.env.getJobId()

mssparkutils.env.getJobId()

Env.GetJobId()

mssparkutils.env.getWorkspaceName()

mssparkutils.env.getWorkspaceName()

mssparkutils.env.getWorkspaceName()

Env.GetWorkspaceName()

mssparkutils.env.getPoolName()

mssparkutils.env.getPoolName()

mssparkutils.env.getPoolName()

Env.GetPoolName()

mssparkutils.env.getClusterId()

mssparkutils.env.getClusterId()

mssparkutils.env.getClusterId()

Env.GetClusterId()

mssparkutils.runtime.context

ctx <- mssparkutils.runtime.context()
for (key in ls(ctx)) {
    writeLines(paste(key, ctx[[key]], sep = "\t"))
}

%%spark
mssparkutils.runtime.context

mssparkutils.session.stop()

mssparkutils.session.stop()

mssparkutils.session.stop()

PyPI 包

Cran 包

Maven 依赖项