主页 / coding / 问题

问题[pyspark](coding)

Martin Hope
mabanalyst
Asked: 2025-01-19 04:45:15 +0800 CST
  • 5

目前，我正在对包含借款人如何偿还贷款的信息的数据库进行计算。

从技术角度来看，我正在使用 PySpark，刚刚面临如何使用高级过滤操作的问题。

我的数据框如下所示：

ID     ContractDate LoanSum  ClosingDate Status Bank
ID3    2024-06-10   20                   Active A
ID3    2024-06-11   30                   Active A
ID3    2024-06-12   50                   Active A
ID3    2024-06-12   15                   Active B
ID3    2024-06-12   5        2024-06-18  Closed A
ID3    2024-06-13   40       2024-06-20  Closed A
ID3    2024-06-22   50                   Active A
ID4    2024-07-11   20                   Active A
ID4    2024-07-12   30                   Active B
ID4    2024-07-13   50                   Active B
ID4    2024-07-11   5        2024-08-20  Closed A

我的目标是通过“贷款金额”字段计算借款人的总和，这些借款人拥有同一家银行在首笔贷款发放之日起 3 天内发放的 3 笔或更多有效贷款。

在我的例子中，ID3 的总和为 20 + 30 + 50 = 100

我目前所做的：

from pyspark.sql import functions as f
from pyspark.sql import Window

df = spark.createDataFrame(data).toDF('ID','ContractDate','LoanSum','ClosingDate', 'Status', 'Bank')
df.show()

cols = df.columns
w = Window.partitionBy('ID').orderBy('ContractDate')

df.withColumn('PreviousContractDate', f.lag('ContractDate').over(w)) \
  .withColumn('Target', f.expr('datediff(ContractDate, PreviousContractDate) >= 0 & datediff(ContractDate, PreviousContractDate) <= 3')) \
  .withColumn('Target', f.col('Target') | f.lead('Target').over(w)) \
  .filter('Target == True')

此代码仅有助于根据 ContractDate 获取向一名借款人发放的贷款。

我如何添加更多条件？

pyspark
Martin Hope
Danylo Kuznetsov
Asked: 2025-01-07 18:37:53 +0800 CST
  • 5

我是 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
Martin Hope
telecomshy
Asked: 2024-12-28 19:51:47 +0800 CST
  • 6

我已经在笔记本电脑上安装了 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
Martin Hope
LearneR
Asked: 2024-12-19 15:26:24 +0800 CST
  • 5

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

is_valid != false

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

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

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

pyspark
Martin Hope
MetallicPriest
Asked: 2024-11-08 21:44:29 +0800 CST
  • 5

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

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

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

pyspark
Martin Hope
user1783504
Asked: 2024-11-01 01:16:02 +0800 CST
  • 6

我正在尝试使用 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
Martin Hope
Pachu Martinez
Asked: 2024-09-04 22:32:47 +0800 CST
  • 5

尝试使用 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
Martin Hope
elokema
Asked: 2024-08-30 23:08:48 +0800 CST
  • 5

我想在函数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'))
pyspark
Martin Hope
1zftqITss0
Asked: 2024-08-27 18:28:06 +0800 CST
  • 5

我在 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
Martin Hope
Mikhail Genkin
Asked: 2024-06-04 08:19:03 +0800 CST
  • 6

在我的 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