问题[apache-spark](coding)

我想创建一个新列，其中包含列中列出的列名的值数组lookup。

示例输入

input_df = spark.createDataFrame([
    Row(id=123, alert=1, operation=1, lookup=[]),
    Row(id=234, alert=0, operation=0, lookup=['alert']),
    Row(id=345, alert=1, operation=0, lookup=['operation']),
    Row(id=456, alert=0, operation=1, lookup=['alert', 'operation']),
])

预期输出

我尝试过

input_df.withColumn("lookup_values", F.transform(F.col("lookup"), lambda x: input_df[f'{x}'])).show()

失败并出现以下错误：

AnalysisException：[UNRESOLVED_COLUMN.WITH_SUGGESTION] 无法解析名称为 的列或函数参数Column<'x_1'>。您是指下列之一吗？[ id, alert, operation, lookup]。

这个错误令人惊讶，因为下面的代码虽然没有产生预期的结果，但却没有产生错误：

input_df.withColumn("lookup_values", F.transform(F.col("lookup"), lambda x: input_df['alert'])).show()

当将多个镶木地板文件读入数据框时，它似乎会随后对每个镶木地板文件进行评估以进行后续转换，而它应该对数据框进行评估。

我正在使用 pyspark 在 fabric notebook 中工作。我试图将多个 parquet 文件读入一个数据框。每个 parquet 文件的列数相同，但列架构可能不同，例如，一个名为“adjustment”的列可能是 int 类型，但如果留空，则类型为 string。我目前正在将文件读入我的数据框，如下所示

df = spark.read.schema(schema).parquet(*files).withColumn(
    "file_name", split(input_file_name(), "/").getItem(8)
)

我在其中指定了架构，文件是我想要从我的湖中加载的文件的文件路径列表。File_name 只是包含日期的文件的名称。

当我跑步时

display(df.where(col("file_name").contains("2024-10-01")))

它似乎可以很好地显示数据框，类似于 display(df)，但是当我运行

display(df.where(col("file_name").contains("2024-12-01")))

它给了我这个错误

org.apache.spark.SparkException: Parquet column cannot be converted in file abfss://[email protected]/lakehouse/path/to/my/data/Data_2024-12-01. Column: [data.adjustment], Expected: string, Found: INT64.

我尝试过指定架构，尝试过 .cache() 或 .persist() 数据框，但每次都出现此错误。我认为这与惰性求值有关，但除了分别读取每个 parquet 文件，然后在对每列强制执行架构更改后将它们合并之外，我实在想不出还能做什么。在此先感谢您的帮助

我最近在使用 Hive 2.3.9 版本时遇到了一个问题。

当我的分组依据设置为“a，b，c”并且分组集设置为（a，b，（a，b））时。

在上一个 Hive 版本中，grouping__id 的结果是当 "a" 时为 1 当 "b" 时为 2 当 "a,b" 时为 3。但是在 2.3.9 版本中，我发现结果是当 "a" 时为 1 当 "b" 时为 2 当 "a, b" 时为 0。新的正则与 Spark grouping__id Regular 相同。

我记得Hive Grouping__id的规则是从低到高，默认都是0，当出现维度时就重置为1，最后进行小数化。

但现在看上去好像已经改变了。

那么 Hive 做出了一些改变吗？

我只是想将 Rate 与 Structured Streaming 结合使用，以便将每个 MicroBatch 写入多个表名。即在 pyspark 中刷新多个接收器逻辑以准备某些认证。

没有错误，但没有发生持久性。我看的时候已经有一段时间了；一定是一些基本的东西。

在 Databricks 社区版上进行如下编码，没有 Hive 目录。基本内容。

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, concat, lit

spark = SparkSession.builder \
    .appName("SimulateKAFKAandMultipleSinks") \
    .getOrCreate()

rate_stream = spark.readStream \
    .format("rate") \
    .option("rowsPerSecond", 1) \
    .load()

message_stream = rate_stream.select(
    (rate_stream["value"] + 1000).alias("message_id"),              
    rate_stream["timestamp"].alias("event_time"),                    
    (concat(lit("T"), rate_stream["value"] % 5)).alias("table_id")   
)

def append_to_parquet(df, table_id):
    table_path = f"/mnt/parquet/{table_id}"   

    df.write \
        .format("parquet") \
        .mode("append") \
        .option("path", table_path) \
        .save()

def process_batch(df, batch_id):
    partitioned_df = df.repartition("table_id")
    def process_partition(iterator):
        for partition in iterator:
            first_row_value = df.first()
            table_id_value = first_row_value['table_id']
            print(f"Writing partition for table_id: {table_id_value}") 
            partition_df = partition.filter(col("table_id") == table_id_value)
            append_to_parquet(partition_df, table_id_value)

    partitioned_df.rdd.mapPartitions(process_partition)

query = message_stream.writeStream \
    .foreachBatch(process_batch) \
    .outputMode("append") \
    .option("checkpointLocation", "/mnt/parquet/checkpoints/") \
    .start()

query.awaitTermination()

更新。我这边有疏忽，但足够有教育意义。可以使用批处理 KAFKA 来实现，而不是通过结构化流。

我有一个如下的数据框：

f1     |f2
=========
test   | [{"f3": 1, "f4": "f4_1" }, {"f3": 2, "f4": "f4_2" }] 

f2是对象列表

我想要获得如下所示的数据框：

f3|f4    | temp_col
=========================
1 |"f4_1"| {"f1": "test"}
2 |"f4_2"| {"f1": "test"}

temp_col是我提供的名字。

我如何使用 pyspark 来实现这一点？

我曾尝试json_normalize通过转换为 pandas df 来使用，但没有效果。

我有一个带有如下数组列的数据集：

+--------+--------------------+
|user_id |array_of_items      |
+--------+--------------------+
|4049494 |[x1, x2, x3, x4, x5]|
|3046908 |[x6, x7, x8, x9]    |
...

如何在不使用 udf 的情况下仅使用 SQL 来查找如下所示的数组？

+--------+----------------------------+
|user_id |array_of_items_diff         |
+--------+----------------------------+
|4049494 |[x2-x1, x3-x2, x4-x3, x5-x4]|
|3046908 |[x7-x6, x8-x7, x9-x8]       |
...

我正在研究PySpark 的聚合函数这个示例Window。

这是数据框：

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('SparkByExamples.com').getOrCreate()
simpleData = (("James", "Sales", 3000), \
    ("Michael", "Sales", 4600),  \
    ("Robert", "Sales", 4100),   \
    ("Maria", "Finance", 3000),  \
    ("James", "Sales", 3000),    \
    ("Scott", "Finance", 3300),  \
    ("Jen", "Finance", 3900),    \
    ("Jeff", "Marketing", 3000), \
    ("Kumar", "Marketing", 2000),\
    ("Saif", "Sales", 4100) \
  )
columns= ["employee_name", "department", "salary"]
df = spark.createDataFrame(data = simpleData, schema = columns)
+-------------+----------+------+
|employee_name|department|salary|
+-------------+----------+------+
|James        |Sales     |3000  |
|Michael      |Sales     |4600  |
|Robert       |Sales     |4100  |
|Maria        |Finance   |3000  |
|James        |Sales     |3000  |
|Scott        |Finance   |3300  |
|Jen          |Finance   |3900  |
|Jeff         |Marketing |3000  |
|Kumar        |Marketing |2000  |
|Saif         |Sales     |4100  |
+-------------+----------+------+

教程中的一个 WindowSpec 按“部门”对行进行分区，并按每个部门内的“薪水”排序：

from pyspark.sql.window import Window
windowSpec  = Window.partitionBy("department").orderBy("salary")

为了熟悉Window操作，我尝试添加一个列“MaxRowNum”，其中包含每个分区内的最大行数。为了方便起见，中间列“RowNum”包含每个分区内的行数：

from pyspark.sql.functions import row_number
df \
.withColumn('RowNum',row_number().over(windowSpec)) \
.withColumn('MaxRowNum',max(col('RowNum')).over(windowSpec)) \
.show()
+-------------+----------+------+------+---------+
|employee_name|department|salary|RowNum|MaxRowNum|
+-------------+----------+------+------+---------+
|        Maria|   Finance|  3000|     1|        1|
|        Scott|   Finance|  3300|     2|        2|
|          Jen|   Finance|  3900|     3|        3|
|        Kumar| Marketing|  2000|     1|        1|
|         Jeff| Marketing|  3000|     2|        2|
|        James|     Sales|  3000|     1|        2|
|        James|     Sales|  3000|     2|        2|
|       Robert|     Sales|  4100|     3|        4|
|         Saif|     Sales|  4100|     4|        4|
|      Michael|     Sales|  4600|     5|        5|
+-------------+----------+------+------+---------+

如上所示，“RowNum”值是正确的，但“MaxRowNum”不包含每个分区内的最大行数。它们仅包含行数，但绑定行除外，绑定行包含两个行数中较大的一个。

从本教程的后面部分开始，我发现了一个没有排序的 WindowSpec，它给了我正确的结果（参见“MaxRowCORRECT”列）：

windowSpecAgg  = Window.partitionBy("department") # No sorting

df.withColumn("row",row_number().over(windowSpec)) \
  .withColumn('MaxRowNum',max(col('row')).over(windowSpec)) \
  .withColumn("MaxRowCORRECT",max(col("row")).over(windowSpecAgg)) \
  .show()
+-------------+----------+------+---+---------+-------------+
|employee_name|department|salary|row|MaxRowNum|MaxRowCORRECT|
+-------------+----------+------+---+---------+-------------+
|        Maria|   Finance|  3000|  1|        1|            3|
|        Scott|   Finance|  3300|  2|        2|            3|
|          Jen|   Finance|  3900|  3|        3|            3|
|        Kumar| Marketing|  2000|  1|        1|            2|
|         Jeff| Marketing|  3000|  2|        2|            2|
|        James|     Sales|  3000|  1|        2|            5|
|        James|     Sales|  3000|  2|        2|            5|
|       Robert|     Sales|  4100|  3|        4|            5|
|         Saif|     Sales|  4100|  4|        4|            5|
|      Michael|     Sales|  4600|  5|        5|            5|
+-------------+----------+------+---+---------+-------------+

我的理解是，窗口聚合函数对每个分区的整体进行操作。上面的代码表明情况不一定如此。我浏览了 Windows 文档，但找不到这种条件行为的明确描述。

是否有一致且完整记录的 Windows 函数操作方案？我在文档中的哪个地方遗漏了它？

背景

根据mazaneicha 的回答，我意识到我需要了解窗口函数分类的背景知识。PySpark 链接到相关术语会产生空白页（currentRow、unboundedPreceding、 unboundedFollowing）。这些东西似乎来自 SQL。虽然我没有在rowFrame和上找到任何内容，但以下页面提供了上述其他术语的背景知识（ rowsBetweenrangeFrame的文档也是如此）：

• https://mode.com/sql-tutorial/sql-window-functions
• https://www.geeksforgeeks.org/window-functions-in-sql
• https://www.databricks.com/blog/2015/07/15/introducing-window-functions-in-spark-sql.html
• 如何在 PySpark 中的 rowsBetween 中使用 unboundedPreceding、unboundedFollowing 和 currentRow

我有一个镶木地板文件，该文件正在通过如下连续循环进行写入：

    def process_data(self):
        #... other code ...

        with pq.ParquetWriter(self.destination_file, schema) as writer:
            with tqdm(total=total_rows, desc="Processing nodes") as pbar:
                for i in range(0, total_rows, self.batch_size):
                    # ... processing code ...

                    # Create a table from the batched data
                    batch_table = pa.Table.from_arrays(
                        [
                            pa.array(node_ids),
                            pa.array(mut_positions),
                            pa.array(new_6mers),
                            pa.array(context_embeddings),
                            pa.array(nonmutation_contexts),
                        ],
                        schema=schema
                    )

                    # Write the batch table
                    writer.write_table(batch_table)

                    # ...

                    pbar.update(len(batch_indices))

由于计算机在此过程中突然关闭，此循环被突然切断。

现在，当我尝试读取文件时pq.read_table，我（预计）收到一个错误

pyarrow.lib.ArrowInvalid: Error creating dataset. Could not read schema from 'data/processed/data_with_embeddings.parquet'. Is this a 'parquet' file?: Could not open Parquet input source 'data/processed/data_with_embeddings.parquet': Parquet magic bytes not found in footer. Either the file is corrupted or this is not a parquet file.

我迫切希望有办法解决这个问题。比如一种变通方法。比如丢失几行（或更多行），但保存大部分数据。我搜索了网络，但似乎没有关于此的任何信息，或者现有的信息超出了我的专业知识范围（这可能从我的标签使用中可以看出，我提前道歉）。

还有希望吗？

如何访问 PySpark 应用程序中预留的用户内存？

我猜测这在 PySpark 应用程序中是不可能的，因为这是 JVM 内存的一部分，无法通过 Python 访问。

如果我是正确的，那么在 PySpark 应用程序中不需要为此留出任何内存（因为它不可访问）？

我有一个时间序列数据集。我希望创建一个新列来表示最后报告的（非空）值。我想我已经弄清楚了这部分，使用lag和的组合last

我还想知道最后报告的（非空）值的时间戳。我从不希望timestamp_ms它是空的，尽管它val可以是空的。

示例数据

df = spark.createDataFrame([
    Row(timestamp_ms=1672531200000, val='19'),
    Row(timestamp_ms=1672532100000, val='20'),
    Row(timestamp_ms=1672533000000, val=None),
    Row(timestamp_ms=1672533900000, val='22'),
    Row(timestamp_ms=1672534800000, val=None),
    Row(timestamp_ms=1672535700000, val=None),
    Row(timestamp_ms=1672536600000, val='25'),
    Row(timestamp_ms=1672537500000, val='20'),
    Row(timestamp_ms=1672538400000, val='27')
])
df.show()

示例代码

返回最后的滞后值并尝试返回报告该值的时间戳。

df_lag = df.withColumn("lag_prev_val", F.lag("val")\
                                    .over(Window.partitionBy()\
                                    .orderBy("timestamp_ms"))
                         )\
            .withColumn("last_lag_prev_val", F.last("lag_prev_val", True)\
                                    .over(Window.partitionBy()\
                                    .orderBy("timestamp_ms"))
                         )\
            .withColumn("last_lag_prev_time", F.lag("timestamp_ms")\
                                .over(Window.partitionBy()\
                                .orderBy("timestamp_ms"))
                     )
df_lag.show()

电流输出

last_lag_prev_time表示先前报告的时间戳，而不是与last_lag_prev_val

理想输出

我想要的输出（加粗差异）是用于last_lag_prev_time表示与用于填充“last_lag_prev_val”的timestamp_ms原始值来自同一行的列val