可以使用正则表达式提取dbname和表：

val result = df.select(
    col("table_name"),
    regexp_replace(col("table_name"), "\\.[^,]+(,|$)", "$1").as("DBName"),
    regexp_replace(col("table_name"), "(^|,)[^,]+\\.", "$1").as("Table")
)

result.show(false)
+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------+
|table_name                                                                                                                                                                                                                                                           |DBName                                                                                                  |Table                                                                                                                                                       |
+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------+
|tradingpartner.parent_supplier,lookup.store,lab_promo_invoice.tl_cc_mbr_prc_wkly_inv,lab_promo_invoice.mpp_club_card_promotion_funding_view,lab_promo_invoice.supplier_sale_apportionment_cc,tradingpartner.supplier,stores.rpm_zone_location_mapping,lookup.calendar|tradingpartner,lookup,lab_promo_invoice,lab_promo_invoice,lab_promo_invoice,tradingpartner,stores,lookup|parent_supplier,store,tl_cc_mbr_prc_wkly_inv,mpp_club_card_promotion_funding_view,supplier_sale_apportionment_cc,supplier,rpm_zone_location_mapping,calendar|
+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------+

抱歉，我用java为您的需求编写了udf，但是我认为它很容易转换成scala。

//Spark >= 2.3
UserDefinedFunction splitTableNameUDF = udf(transformTables(), getTableName());
df.withColumn("table_name_new", splitTableNameUDF.apply(col("table_name")))
    .select("table_name", "table_name_new.DBName", "table_name_new.Table");

//Spark < 2.3

sqlContext.udf().register("splitTableNameUDF", transformTables(), getTableName());
df.withColumn("table_name_new", callUDF("splitTableNameUDF", col("table_name")))
    .select("table_name", "table_name_new.DBName", "table_name_new.Table");

//schema

public static StructType getTableName() {
    List<StructField> inputFields = new ArrayList<>();
    inputFields.add(DataTypes.createStructField("DBName", DataTypes.StringType, true));
    inputFields.add(DataTypes.createStructField("Table", DataTypes.StringType, true));
    return DataTypes.createStructType(inputFields);
}

//UDF

public static UDF1<String, Row> transformTables() {
    return (row) -> {
        String[] dbtableNames = row.split(",");
        List<String> dbNames = new ArrayList<>();
        List<String> tableNames = new ArrayList<>();
        for (String dbtableName : dbtableNames) {
            String[] split = dbtableName.split(".");
            dbNames.add(split[0]);
            tableNames.add(split[1]);
        }
        return RowFactory.create(String.join(",", dbNames), String.join(",", tableNames));
    };
}