问题[pyspark](coding)

我是 PySpark 的新手，目前在 Databricks 中工作，比较两个具有相同列结构的 DataFrame。我将它们相互比较（本质上是将已加载到数据库中的文件与新文件进行比较）。在此过程中，我使用以下代码计算对每个变量所做的更改次数：

Comparison_DF = DF1_Data_To_Compare.withColumn("Value1_Change", when(col("b.Value1") == col("a.Value1"), 0).otherwise(1))
Comparison_DF = Comparison_DF.withColumn("Value2_Change", when(col("b.Value2") == col("a.Value2"), 0).otherwise(1))  

# Summarizing the number of changes
Change_To_Value1 = Comparison_DF.select(sum("Value1_Change"))
Change_To_Value2 = Comparison_DF.select(sum("Value2_Change"))

# Forming the change report DataFrame
# columns=["Type of Change", "Number of Occurrences"]
data = [("Change to Value1", Change_To_Value1), ("Change to Value2", Change_To_Value2)]

rdd = spark.sparkContext.parallelize(data)
print(data) 

该行rdd = spark.sparkContext.parallelize(data)返回错误。检查错误回溯后，我意识到Change_To_Value1和Change_To_Value2不是变量而是 DataFrames。该print(data)语句给出以下结果：[('Change to Value1', DataFrame[sum(Value1_Change): bigint]), ('Change to Value2', DataFrame[sum(Value2_Change): bigint])]。

我需要形成这种 DataFrame 以将其用作更改报告，以便与 SSIS 包返回的结果进行比较。

我在 StackOverflow 或任何其他开源上都没有找到类似的东西。我尝试构建一个循环语句来收集这些 DataFrame 并直接将它们输入到新的 DataFrame 中，但我也失败了。

有没有办法将这些 DataFrame 转换为 int 变量？或者有没有更好的方法来形成这个 DataFrame？

我已经在笔记本电脑上安装了 PySpark。当我运行以下程序时，一切正常：

spark = SparkSession.builder.appName('pyspark').getOrCreate()
book_local = spark.read.text("data.txt")
book_local.show()

但是，当我运行以下程序时，出现错误：

spark = SparkSession.builder.appName('pyspark').getOrCreate()

my_grocery_list = [
    ["Banana", 2, 1.74],
    ["Apple", 4, 2.04],
    ["Carrot", 1, 1.09],
    ["Cake", 1, 10.99],
]
df_grocery_list = spark.createDataFrame(my_grocery_list)
df_grocery_list.show()   # This is where the error is thrown

错误信息是：

Py4JJavaError: java.io.IOException: Cannot run program "python3"

设置环境变量后，一切恢复正常。

import os
import sys
from pyspark.sql import SparkSession

os.environ['PYSPARK_PYTHON'] = sys.executable
os.environ['PYSPARK_DRIVER_PYTHON'] = sys.executable

我的问题是，为什么第一个程序运行没有问题，但第二个程序却抛出了 Py4J 错误？第一个程序根本没有使用 Py4J 包吗？

此外，当我尝试用以下代码替换环境变量配置时：

spark = SparkSession.builder.appName('pyspark').config("spark.pyspark.python", sys.executable).getOrCreate()

我仍然遇到错误。

我的 PySpark 笔记本中有一个TempView。当我在视图上运行 SQL 查询时，WHERE以下条件不会产生预期的结果：

is_valid != false

对于传入的记录，is_valid列（BooleanType字段）包含一个null值。我期望条件is_valid != false计算为，true因为null显然不等于false，对吗？

然而，事实并非如此。相反，它的计算结果为false。

我不确定我的情况出了什么问题。有什么想法吗？

我有一个从 Databricks 访问的增量表。我有一个结构体数组类型的列。如果我想查看数组中任何元素的字段是否包含某个元素，我可以使用以下命令。

array_contains(transform(ArrayCol, x -> x.f), 'something')

但是现在我不想进行完全匹配，而是进行类似于的操作LIKE 'some%'。我该如何实现？基本上，我想返回true数组中的任何元素是否具有f包含此处模式的字段'some%'。

我正在尝试使用 pyspark 中的多维数据集函数，但不包含多维数据集中的所有列。

我想要实现的 SQL 等效目标：

从表中按 col1、col2、col3、sum(col4) 分组，选择 col1、col2、col3

这将按 col1 以及 col2 和 col3 的所有组合对组进行分组

在 pyspark 中，运行以下命令时，我收到消息 GroupedData 对象没有属性“cube”

spark.table("table").groupBy(col1).cube(col2,col3).agg(sum(col4))

我可以使用 cube，但是我需要包含我不想要的 col1

spark.table("table").cube(col1,col2,col3).agg(sum(col4))

尝试使用 pyspark 从 JDBC读取。在 JDBC 中，有一列 FEC_PART 作为日期类型，格式为 yyyymmdd。对于读取，参数upperBound或lowerBound与所需格式yyyymmdd不匹配：

• 使用所需的格式yyyymmdd，出现无法识别日期格式的错误： pyspark.sql.utils.IllegalArgumentException: Cannot parse the bound value 20200112 as date
• 格式为yyyy-mm-dd，与 JDBC 中 FEC_PART 的格式不匹配。它显示此错误 java.sql.SQLDataException: ORA-01861: literal does not match format string： WHERE "ARQPIB_FEC_PART" < '2020-01-13' or "ARQPIB_FEC_PART" is null , Error Msg = ORA-01861: literal does not match format string

使用spark进行read.load()时出现错误。

input_data = spark.read \
    .format(constants.FORMAT_JDBC) \
    .options(**properties) \
    .option("partitionColumn", "FEC_PART")  # Keep partition column as it is
    .option("lowerBound", "20200112")  # Use the yyyymmdd format for bounds to match partitionColumn
    .option("upperBound", "20200114")  # Use the yyyymmdd format for bounds to match partitionColumn
    .option("numPartitions", "2") \
    .load()

第一种方法。尝试添加此选项：

.option("oracle.jdbc.mapDateToTimestamp", "false")
.option("sessionInitStatement", "ALTER SESSION SET NLS_DATE_FORMAT = 'YYYYMMDD'")

采用另一种选择的第二种方法：

.option("dateFormat", "yyyyMMdd")

又尝试了一些方法，但都没有任何结果。

我想在函数pyspark中写入这部分

df = (df.withColumn("January", F.lit(None).cast('double'))
        .withColumn("February", F.lit(None).cast('double'))
        .withColumn("March", F.lit(None).cast('double'))
        .withColumn("April", F.lit(None).cast('double'))
        .withColumn("May", F.lit(None).cast('double'))
        .withColumn("June", F.lit(None).cast('double'))
        .withColumn("July", F.lit(None).cast('double'))
        .withColumn("August", F.lit(None).cast('double'))
        .withColumn("September", F.lit(None).cast('double'))
        .withColumn("November", F.lit(None).cast('double'))
        .withColumn("December", F.lit(None).cast('double'))

我在 Microsoft Fabric 中使用 Spark Notebook。我想从 Lakehouse 中的元数据构建列映射。该映射应写入带有表列表的数据框中的“映射”列中。

我目前的尝试如下：

# Create initial list of table data
dataframe_tablelist = spark.createDataFrame(
    [
        ("abcd", "AB", "t1"),
        ("efgh", "CD", "t2"),
        ("efgh", "CD", "t3"),
    ],
    ["database", "entity", "table_name"]
)

def construct_mapping(database, entity, table_name):
    meta_name = "Metadata_" + database + "_" + entity + "_" + table_name
    metadata = spark.sql(f"""select * from {meta_name}""")
    # Here I would construct the mapping from the metadata
    return meta_name

udf_constructor = udf(construct_mapping, StringType())

mapping_df = dataframe_tablelist.withColumn("test_column", udf_constructor(dataframe_tablelist.database, dataframe_tablelist.entity, dataframe_tablelist.table_name))

display(mapping_df)

我收到了这个我完全不明白的错误：

PicklingError: Could not serialize object: PySparkRuntimeError: [CONTEXT_ONLY_VALID_ON_DRIVER] It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transformation. SparkContext can only be used on the driver, not in code that it run on workers. For more information, see SPARK-5063.

我可能可以让它与 collect() 一起工作并逐行附加，但我想以“正确”的方式进行。

在我的 pyspark 作业中，我有一个巨大的数据框架，其中有超过 6,000 列，格式如下：

id_        a1   a2     a3   a4     a5   .... a6250
u1827s    True False  True False False .... False
...

其中大多数列a1,a2,a3,...,a6250 都是二进制类型。我需要按所有这些列对这些数据进行分组，并聚合每个组合的不同 id 的数量，例如

df = df.groupby(list_of_cols).agg(F.countDistinct("id_"))

在哪里list_of_cols = [a1,a2,...,a6250]。运行此 pyspark 作业时，出现java.lang.StackOverflowError错误。我知道我可以增加堆栈大小（按照https://rangereddy.github.io/SparkStackOverflow/），但是，我更喜欢一个更优雅的解决方案，也可以实现更方便的输出。

在分组之前我有两个想法：

1. 将 a1,a2,...,a6250 列的组合编码为单个二进制列，例如具有 6250 位的二进制数，其中位置上的位k将为该列编码 True 或 False 值a_k，例如在上面的示例中的值将是10100...0（a1 为真，a2 为假，a3 为真，a4 为假，a5 为假，...a6250 为假）。

2. 将这些值收集到一个二进制数组中，例如有 1 列，如数组（True，False，True，False，False，....，False）。

哪种方法更好 - 增加堆栈大小并处理 6000 多个列、使用单个二进制列还是二进制值数组？

我有一个 pyspark DataFrame，其中包含列，每个列都是一个字符串数组，如何创建一个新列，它是它们的笛卡尔积，而不将它们拆分为两个数据帧并连接它们，并且没有 udf？

例子：

In df:
Df
+---+---+---+---+-
| a1    | a2     |
+---+---+---+---+-
|[1, 2]|[3, 4, 5]|
|[1, 2]|[7, 8]   |
+---+---+---+---+-

Out df:
+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
| a1    | a2     | a3                                               |
+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
|[1, 2]|[3, 4, 5]|[{1, 3}, {1, 4}, {1, 5}, {2, 3}, {2, 4}, {2, 5}]  |
|[1, 2]|[7, 8]   |[{1, 7}, {1, 8}, {2, 7}, {2, 8}]                  |
+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+