在Databricks中使用Scala实现工作流任务间的信息共享

嗨，我来帮你搞定这个问题！其实在Scala任务里，你完全可以直接用和Python类似的方式调用dbutils的API来实现任务间的信息传递，另外我也会给你几个实用的替代方案，供你根据场景选择。

一、直接使用Scala版的dbutils.jobs.taskValues #

Databricks的dbutils工具在Scala环境里同样可用，语法和Python非常接近，只是适配了Scala的语法规则：

1. 在前置任务中设置共享值 #

在第一个Scala任务里，你可以这样设置要传递的键值对：

// 简洁写法
dbutils.jobs.taskValues.set("name", "Some User")
// 更清晰的命名参数写法
dbutils.jobs.taskValues.set(key = "name", value = "Some User")

2. 在后续任务中获取共享值 #

在第二个Scala任务里，通过指定前置任务的taskKey来获取值，还可以设置默认值防止获取失败：

// 获取值，注意返回类型是Any，需要转成你需要的类型（比如String）
val userName = dbutils.jobs.taskValues.get(
  taskKey = "prev_task_name", 
  key = "name", 
  default = "Jane Doe"
).asInstanceOf[String]

// 直接使用变量
println(s"Hello, $userName!")

如果传递的是复杂类型（比如Map、自定义Case Class），建议先序列化成JSON字符串传递，再在接收端反序列化，避免类型转换出现问题。

二、替代方案：适合复杂数据或特殊场景 #

如果因为某些限制（比如旧版本Databricks对Scala的taskValues支持有问题），你可以试试这些替代方法：

1. 使用Delta表存储共享数据 #

如果要传递的数据量较大或结构复杂，可以在第一个任务里把数据写入Delta表，第二个任务直接读取：

// 前置任务：写入数据到Delta表
val sharedData = Seq(("Some User", 30)).toDF("name", "age")
sharedData.write.mode("overwrite").saveAsTable("temp.shared_user_data")

// 后续任务：读取数据
val userData = spark.read.table("temp.shared_user_data")
val userName = userData.select("name").head().getString(0)

这种方法跨集群、跨任务都能稳定使用，比临时视图更可靠（临时视图仅在同一集群会话中有效）。

2. 使用DBFS存储文件 #

把数据序列化成JSON、Parquet等格式写入DBFS，后续任务读取解析：

// 前置任务：写入JSON到DBFS
import org.json4s._
import org.json4s.jackson.Serialization._
import java.io.FileWriter

implicit val formats = DefaultFormats
val userInfo = Map("name" -> "Some User", "email" -> "user@example.com")
val jsonStr = write(userInfo)

val writer = new FileWriter("/dbfs/tmp/shared_user_info.json")
writer.write(jsonStr)
writer.close()

// 后续任务：读取并解析JSON
import scala.io.Source
val jsonContent = Source.fromFile("/dbfs/tmp/shared_user_info.json").mkString
val parsedInfo = parse(jsonContent).extract[Map[String, String]]
val userName = parsedInfo("name")

3. 使用Databricks Secrets（适合敏感数据） #

如果要传递的是敏感信息（比如密钥、密码），可以在前置任务中把值写入Databricks Secrets，后续任务读取：

// 前置任务：写入Secret
dbutils.secrets.put(scope = "my-secrets-scope", key = "shared-user-name", stringValue = "Some User")

// 后续任务：读取Secret
val userName = dbutils.secrets.get(scope = "my-secrets-scope", key = "shared-user-name")

这个方法仅适合小体量的敏感数据，不适合传递普通业务数据。

备注：内容来源于stack exchange，提问作者stuffed