我有一个带有如下数组列的数据集：
+--------+--------------------+
|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的文档也是如此）：
我有一个镶木地板文件，该文件正在通过如下连续循环进行写入：
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
| 时间戳毫秒 | 瓦尔 | lag_prev_val | last_lag_prev_val | last_lag_prev_time |
|---|---|---|---|---|
| 1672531200000 | 19 | 无效的 | 无效的 | 无效的 |
| 1672532100000 | 20 | 19 | 19 | 1672531200000 |
| 1672533000000 | 无效的 | 20 | 20 | 1672532100000 |
| 1672533900000 | 22 | 无效的 | 20 | 1672533000000 |
| 1672534800000 | 无效的 | 22 | 22 | 1672533900000 |
| 1672535700000 | 无效的 | 无效的 | 22 | 1672534800000 |
| 1672536600000 | 二十五 | 无效的 | 22 | 1672535700000 |
| 1672537500000 | 20 | 二十五 | 二十五 | 1672536600000 |
| 1672538400000 | 二十七 | 20 | 20 | 1672537500000 |
理想输出
我想要的输出（加粗差异）是用于last_lag_prev_time表示与用于填充“last_lag_prev_val”的timestamp_ms原始值来自同一行的列val
| 时间戳毫秒 | 瓦尔 | lag_prev_val | last_lag_prev_val | last_lag_prev_time |
|---|---|---|---|---|
| 1672531200000 | 19 | 无效的 | 无效的 | 无效的 |
| 1672532100000 | 20 | 19 | 19 | 1672531200000 |
| 1672533000000 | 无效的 | 20 | 20 | 1672532100000 |
| 1672533900000 | 22 | 无效的 | 20 | 1672532100000 |
| 1672534800000 | 无效的 | 22 | 22 | 1672533900000 |
| 1672535700000 | 无效的 | 无效的 | 22 | 1672533900000 |
| 1672536600000 | 二十五 | 无效的 | 22 | 1672533900000 |
| 1672537500000 | 20 | 二十五 | 二十五 | 1672536600000 |
| 1672538400000 | 二十七 | 20 | 20 | 1672537500000 |
我有这个数据框：
+---------+
| data|
+---------+
|[a, b, c]|
|[d, e, f]|
|[g, h, i]|
+---------+
以及列名列表 [“first col”, “second col”, “third col”]
我想创建新列来生成以下数据框：
+-----------+-----------+----------+
| first col| second col| third col|
+-----------+-----------+----------+
| a| b| c|
| d| e| f|
| g| h| i|
+-----------+-----------+----------+
我正在绞尽脑汁想如何做到这一点，正确的方法是什么？
给定带有月份和年份的日期值：
03.2020
我尝试将其转换为 DateType，如下所示：
to_timestamp(col("Date"), "MM.yyyy").cast(DateType)
但这却带来了一些我没想到的事情：
2017-03-01
返回值包含额外的“01”（日期值），并重新排序月份和年份。因此预期值为“03.2020”
根据这个建议，我还尝试了：
date_format(col("Date"), "MM.yyyy")
但在这种情况下，该函数返回 null。
我究竟做错了什么 ？
感谢您抽出时间回答这个问题。
我最近正在使用 Spark，我读到它认为 HDFS 中的一个分区 = Spark 中的一个分区。按照这个逻辑，很多情况下我们可能不会使用 HDFS 作为源。因此，如果我们使用 CSV 或任何其他基于文件的格式来读取数据，那么分区是如何分区的，或者更确切地说，由于没有显式分区，因此数据是如何分区的。
当 flink 中以批处理模式执行作业时，fileSink 按并行数生成多个文件，但我只想在一个文件中输出而不更改并行数，我该怎么做？
将parallel 放入5 中，FileSink 的输出中仅得到一个文件。
OutputFileConfig config = OutputFileConfig
.builder()
.withPartPrefix("prefix")
.withPartSuffix(".txt")
.build();
final FileSink<String> sinkfile = FileSink
.forRowFormat(new Path("src/main/resources/output"), new SimpleStringEncoder<String>("UTF-8"))
.withBucketAssigner(new BasePathBucketAssigner<>())
.withOutputFileConfig(config)
.build();
我的 SparkSQL 数据框如下所示：
+-----------+----------+----------+----------+
|a |b |c |d |
+-----------+----------+----------+----------+
| 123| abc| N| 2|
| 123| abc| N| 4|
| 123| abc| X| 3|
| 456| def| K| 1|
| 456| def| X| 4|
+-----------+----------+----------+----------+
我想根据键列删除重复项a，并且如果该键组合存在一行在 column 中具有值的值，则该重复项在 列 中b没有值。Nc Nc
预期输出：
+-----------+----------+----------+----------+
|a |b |c |d |
+-----------+----------+----------+----------+
| 123| abc| N| 2|
| 123| abc| N| 4|
| 456| def| K| 1|
| 456| def| X| 4|
+-----------+----------+----------+----------+
对于组合键a = 123，b = abc存在不止一行 where c = N。因此，我们要删除该组合键 where 的所有行c != N。
对于组合键a = 456，b = def不存在任何行c = N。因此我们不想删除任何行。