来自两个不同表 Pyspark 的数据框中的 CountDistinct答案

【问题标题】：CountDistinct in a dataframe from two different table Pyspark来自两个不同表 Pyspark 的数据框中的 CountDistinct
【发布时间】：2021-10-07 09:48:00
【问题描述】：

我对 pyspark 上的 CountDistinct 有疑问。我有两个连接表，我想显示两个不同表的不同键值的数量。

input_table = spark.read.parquet(path + table_name1)
input_table2 = spark.read.parquet(path + table_name2)

impacted_columns = ctrl \
    .where((F.col("tableName1") == table_name1) & (F.col("tableName2") == table_name2)) \
    .join(key, on=['tableName1', 'tableName2']) \
    .join(table, ctrl["tableName1"] == table['tableName']) \
    .groupby("tableName1", 'tableName2', 'key1', 'key2', "keycolumn") \
    .agg(F.collect_list(F.struct("columnName1", "columnName2")).alias('imp_columns')) \
    .first()

input_table = input_table.select(
    [F.col(c.columnName1).alias("T1_" + c.columnName1) for c in impacted_columns.imp_columns] +
    [F.col(c).alias("T1_" + c) for c in
     list(set().union(impacted_columns.key1.split("-"), impacted_columns.keycolumn.split("-")))]
)

input_table2 = input_table2.select(
    [F.col(c.columnName2).alias("T2_" + c.columnName2) for c in impacted_columns.imp_columns] +
    [F.col(c).alias("T2_" + c) for c in
     list(set().union(impacted_columns.key2.split("-"), impacted_columns.keycolumn.split("-")))]
)

ppp= input_table \
        .join(input_table2, [input_table["T1_" + f] == input_table2["T2_" + s] for (f, s) in
                             zip(impacted_columns.key1.split("-"), impacted_columns.key2.split("-"))],
              "left_anti")
plu= input_table2 \
        .join(input_table, [input_table["T1_" + f] == input_table2["T2_" + s] for (f, s) in
                             zip(impacted_columns.key1.split("-"), impacted_columns.key2.split("-"))],
              "left_anti")
pppc=F.countDistinct(ppp.select(["T1_"+c for c in impacted_columns.key1.split("-")]))
pluc=F.countDistinct(plu.select(["T2_" + c for c in impacted_columns.key2.split("-")]))

但它不起作用，有人知道我的代码有什么问题吗？（它一直有效到 countDistinct）

pppc=F.countDistinct(ppp.select(["T1_"+c for c in impacted_columns.key1.split("-")]))
TypeError: Invalid argument, not a string or column: DataFrame[T1_c: string, T1_p: string] of type <class 'pyspark.sql.dataframe.DataFrame'>. For column literals, use 'lit', 'array', 'struct' or 'create_map' function.

【问题讨论】：

标签： python dataframe pyspark count

【解决方案1】：

您收到错误消息，因为您将数据帧传递给 F.countDistinct。 F.countDistinct 的每个参数都应该是一个列名。

尝试用这个交换最后两行：

ppc = ppp.agg(F.countDistinct(*["T1_"+c for c in impacted_columns.key1.split("-")]))
pluc= plu.agg(F.countDistinct(*["T2_" + c for c in impacted_columns.key2.split("-")]))

然后，您可以按如下方式在驱动程序处收集您的不同计数：

print(ppc.collect()[0][0])
print(pluc.collect()[0][0])

【讨论】：

它给了我以下错误：plu.agg(F.countDistinct(*["T2_"+c for c in Impacted_columns.key2.split("-")])).alias(' onlykey2') TypeError: Invalid argument, not a string or column: DataFrame[count(DISTINCT T1_c, T1_p): bigint] of type 。对于列字面量，使用“lit”、“array”、“struct”或“create_map”函数。
我发布的行的输出是什么（没有别名，在数据帧上不起作用）？。 ppuc.show() 的输出是什么？
ppuc.show() 的输出是一个 df，有 1 列：count(DISTINCT T1_Identifier, T1_Username) 和 2 作为计数（我在测试环境中工作）
好的，太好了，我想这解决了你的问题吧？
是的，我只需要一个命令就可以只选择数字（DF的最后一行）