问题[pyspark](coding)

我有一个从 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}]                  |
+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+

在 Pyspark 中，尝试根据长度数组列“Col1”查找偏移量。不想使用 UDF，因此尝试使用转换来获得解决方案。但面临错误。请建议任何解决方法

Col1            Offset
[3,4,6,2,1]     [[0,3],[4,8],[9,15],[16,18],[19,20]]
[10,5,4,3,2]    [[0,10],[11,16],[17,21],[22,25],[26,28]]

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, explode, expr


 spark = SparkSession.builder \
.appName("Calculate Offset Column") \
.getOrCreate()


 data = [([3.0, 4.0, 6.0, 2.0, 1.0],),
    ([10.0, 5.0, 4.0, 3.0, 2.0],)]


 df = spark.createDataFrame(data, ["Col1"])
 df = df.withColumn("Offsets", 
               f.expr("""transform(Col1, (x, i) -> struct(coalesce(sum(Col1) over (order by i rows between unbounded preceding and current row) - x, 0) as start, 
                                                         sum(Col1) over (order by i rows between unbounded preceding and current row) as end))"""))

错误：运算符 !Window 中的 Col1#454 中缺少解析的属性 i#462 [Col1#454，transform(Col1#454，lambdafunction(struct(start，coalesce((sum(cast(Col1 as double))) windowspecdefinition( lambda i#462 ASC NULLS FIRST、specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) - lambda x#461),cast(0 as double)), end, sum(cast(Col1 as double)) windowspecdefinition (lambda i#462 ASC NULLS FIRST，指定窗口帧(RowFrame，unboundedpreceding$()，currentrow$())))，lambda x#461，lambda i#462，false)) AS Offsets#458]，[lambda i#462 ASC 首先为空]。；

我有一个场景，我想连接 2 个数据帧，但没有一个连接（左、右、内部、除外）能满足我的目的。

这是我的数据框 Dataframe 1 ：

日期 | 身份证 | 业主| ID2 | 价值|

20240101 | 2 | 1 | 3 | 100|

20240110 | 2 | 1 | 3 | 200|

数据框2：

日期 | 身份证 | 业主|

20240111 | 2 | 1 |

20240112 | 2 | 1 |

20240103 | 2 | 1 |

输出数据帧：

日期 | 身份证 | 业主| ID2 | 价值|

20240101 | 2 | 1 | 3 | 100|

20240110 | 2 | 1 | 3 | 200|

20240111 | 2 | 1 | 3 | 0|

20240112 | 2 | 1 | 3 | 0|

20240103 | 2 | 1 | 3 | 0|

虽然 Dataframe1 没有某些日期的值，但 Dataframe2 有，因此输出应包含“值”为 0

请帮忙，提前致谢

我有一个 pyspark 数据框，其中包含一些 ID 数据和 2 个位置列，这些列是用逗号分隔的字符串：

我想分解这些列，使它们成为新行：

我怎样才能实现这个目标？