使用SPARK BQ连接器从DataProc上的Pyspark作业中将标签添加到大查询表中

Question

我正在尝试在Google DataProc群集上使用PY-Spark来运行Spark作业并将结果写入大查询表。

Spark BigQuery连接器文档 - https：//githbithub.com/googleclecloclocdataproc/spark-bigquroc/spark-bigquery-conector

要求在桌子的​​创建期间，大查询表中应存在某些标签。

SPARK BQ连接器没有提供任何添加标签以添加标签以进行写操作的规定，

df.write.format("bigquery") \
    .mode("overwrite") \
    .option("temporaryGcsBucket", "tempdataprocbqpath") \
    .option("createDisposition", "CREATE_IF_NEEDED") \
    .save("abc.tg_dataset_1.test_table_with_labels")

以上命令在背景中创建BigQuery Load作业，该作业将表加载数据。 进一步检查后，大查询负载作业语法本身并不支持与大查询 - 查询作业相反的标签。

是否有任何计划支持以下

1. 的大查询负载工作支持中
2. 标签在Spark BQ连接器的写操作中为标签

的支持。由于在加载/写操作过程中没有任何规定要添加标签

Answer 1

问题在10个月前发布。我不确定当时是否可以使用或最近添加它，但是相同的GitHub文档提供了添加标签的选项：

-BigQueryTableLabel： -
可以用来将标签添加到表中时将标签添加到表中。可以设置多个标签。
（可选）

因此您可以在代码中添加类似的内容： -

spark.conf.set（“ bigquerytablelabel.name”，“ sample_table_name”）

Answer 2

这个问题已经发布了很长时间。但是，请通过以下信息回答。它将帮助开发人员获得有关每个Spark Job的插槽用途。

我已经写了下面的代码，以捕获相对于每个Spark Job的BigQuery插槽使用情况。
您可以在下面使用下面的补丁，其中您可以通过Spark Job执行查询。

 def getDataprocJobId(spark: SparkSession): String = {
    spark.sparkContext.getConf.get(KEY_SPARK_YARN_TAGS, UNKNOWN).split(',')
      .find(_.startsWith(DATAPROC_JOB_PREFIX))
      .map(_.substring(DATAPROC_JOB_PREFIX.length))
      .getOrElse(spark.sparkContext.appName)
  }

val bld = QueryJobConfiguration.newBuilder(sql)
val dataprocjobid: String = getDataprocJobId(spark)
  var dataproc_job_id: String = {
    if (dataprocjobid.startsWith("exp")) {
      val pattern= "exp\\-[a-zA-Z0-9-](.*)(?=-EXPORT)".r
      pattern.findFirstIn(dataprocjobid).toString
    }
    else {
      dataprocjobid
    }
  }

import scala.collection.JavaConverters._
bld.setLabels(Map("dataproc_job_id" -> dataproc_job_id.toLowerCase).asJava)

**Its working fine and I can see the labels inside information_schema table.**

select
 query
,TIMESTAMP_DIFF(start_time, end_time, MINUTE) as diff
,total_bytes_processed
,total_slot_ms
,cache_hit
,lbl.value as dataproc_job_id
from `region-eu.INFORMATION_SCHEMA.JOBS_BY_PROJECT` ,unnest(labels) lbl
where date(creation_time) >='2024-04-19'
and lbl.key='dataproc_job_id' limit 10;

但是，它有一定的局限性。

1)Label keys and values can be no longer than 63 characters
2)can only contain lowercase letters, numeric characters, underscores and dashes. International characters are allowed. 
3)Label keys must start with a letter and each label in the list must have a different key.

对于我们的系统，很少有Spark Jobs的工作ID超过63的限制，因此我们使用Regex从中获取有意义的信息。