如何动态生成带SCIO BigQuery注解的Scala类并解决宏编译问题

我需要动态生成Scala类，用于SCIO与GCP BigQuery之间的类型安全读写操作。输入为数据集、源表名、目标表名及目标字段列表，可通过字符串拼接生成类源码，但使用Scala Toolbox或scala.tools.nsc编译时，均出现**'需启用macro paradise（2.12）或-Ymacro-annotations（2.13）'**的错误——内部编译器无法识别build.sbt中的paradise插件配置。想知道能否为scala.tools.nsc.{Global, Settings}添加所需编译器插件配置，或采用其他方法生成这类带注解的动态类？

目标示例代码 #

import com.spotify.scio.bigquery._
import com.spotify.scio.bigquery.types.BigQueryType

@BigQueryType.fromTable("dataset.SOURCE_TABLE")
class SOURCE_TABLE

@BigQueryType.toTable
case class TARGET_TABLE(id: String, name: String, desc: String)

def main(cmdlineArgs: Array[String]): Unit = {
  val (sc, args) = ContextAndArgs(cmdlineArgs)
  sc.typedBigQuery[SOURCE_TABLE]()  // 从BQ读取
    .map(row => transformation(row)) // 转换为SCollection[TARGET_TABLE]
    .saveAsTypedBigQueryTable(Table.Spec(args("TARGET_TABLE")))  // 保存到BQ
  sc.run()
  ()
}

生成类的源码示例 #

val sourceString =
  s"""
     |import com.spotify.scio.bigquery.types.BigQueryType
     |
     |@BigQueryType.fromTable("$dataset.$SOURCE_TABLE")
     |class $SOURCE_TABLE
     |
   """.stripMargin

val targetString =
  s"""
     |import com.spotify.scio.bigquery.types.BigQueryType
     |
     |@BigQueryType.toTable
     |case class $TARGET_TABLE($fieldDefinitions)
   """.stripMargin

当前尝试的编译代码 #

def compileCode(sources: List[String], classpathDirectories: List[AbstractFile], outputDirectory: AbstractFile): Unit = {
  val settings = new Settings
  classpathDirectories.foreach(dir => settings.classpath.prepend(dir.toString))
  settings.outputDirs.setSingleOutput(outputDirectory)
  settings.usejavacp.value = true
  //***** 
  // 添加macros paradise编译器插件？
  //*****
  val global = new Global(settings)
  val files = sources.zipWithIndex.map { case (code, i) => new BatchSourceFile(s"(inline-$i)", code) }
  (new global.Run).compileSources(files)
}

Scala版本：2.12.17

方法一：为scala.tools.nsc.Settings添加Macro Paradise插件配置 #

Scala 2.12下，BigQueryType注解依赖macro paradise插件，需要手动在编译器设置中指定插件路径和启用参数：

import scala.tools.nsc.{Global, Settings}
import scala.tools.nsc.io.AbstractFile

def compileCode(
  sources: List[String],
  classpathDirectories: List[AbstractFile],
  outputDirectory: AbstractFile,
  macroParadiseJarPath: String // 传入macro paradise插件的jar路径
): Unit = {
  val settings = new Settings
  // 配置基础类路径
  classpathDirectories.foreach(dir => settings.classpath.prepend(dir.toString))
  // 添加macro paradise插件
  settings.plugin.value += macroParadiseJarPath
  // 启用插件功能
  settings.pluginOptions.value += "-P:macroparadise:enable"
  // 输出目录配置
  settings.outputDirs.setSingleOutput(outputDirectory)
  settings.usejavacp.value = true

  val global = new Global(settings)
  val files = sources.zipWithIndex.map { case (code, i) => new global.BatchSourceFile(s"(inline-$i)", code) }
  (new global.Run).compileSources(files)
}

注意事项： #

• 需确保macroParadiseJarPath指向正确的插件jar包，比如本地Maven/Ivy仓库中的路径：~/.ivy2/cache/org.scalamacros/macro-paradise_2.12.17/jars/macro-paradise_2.12.17-2.1.1.jar（版本需与Scala 2.12.17匹配）
• 类路径必须包含scio-bigquery相关依赖，否则编译时会找不到BigQueryType注解类

方法二：用Scala Toolbox配置插件 #

如果使用Scala Toolbox进行动态编译，同样需要在Toolbox初始化时指定插件参数：

import scala.reflect.runtime.universe._
import scala.tools.reflect.ToolBox

// 替换为实际的macro paradise jar路径
val macroParadiseJarPath = "/path/to/macro-paradise_2.12.17-2.1.1.jar"
val tb = runtimeMirror(getClass.getClassLoader).mkToolBox(
  options = s"-Xplugin:$macroParadiseJarPath -P:macroparadise:enable"
)

// 编译生成的源码字符串
val sourceSymbol = tb.compile(tb.parse(sourceString))()
val targetSymbol = tb.compile(tb.parse(targetString))()

方法三：绕过动态编译，使用Scio动态API #

如果动态编译的复杂度太高，可以直接使用Scio提供的非类型安全但更灵活的API，无需生成注解类：

def main(cmdlineArgs: Array[String]): Unit = {
  val (sc, args) = ContextAndArgs(cmdlineArgs)
  // 直接读取BigQuery表为TableRow集合
  sc.bigQueryTable(s"$dataset.SOURCE_TABLE")
    .map(row => {
      // 手动转换为目标结构，构建TableRow
      TableRow(
        "id" -> row.get("id"),
        "name" -> row.get("name"),
        "desc" -> row.get("desc")
      )
    })
    // 写入BigQuery，指定表结构
    .saveAsBigQueryTable(
      Table.Spec(args("TARGET_TABLE")),
      schema = Table.Schema(
        Field.of("id", Field.Type.String),
        Field.of("name", Field.Type.String),
        Field.of("desc", Field.Type.String)
      )
    )
  sc.run()
}

这种方式不需要编译时注解，完全在运行时处理，避免了动态编译的插件配置问题。

内容的提问来源于stack exchange，提问作者kolya_metallist