标签: pyspark-dataframes

我有两个数据框，DF1 和 DF2。DF1 为主，DF2 为增量。来自DF2 的数据应插入DF1 或用于更新DF1 数据。

假设 DF1 具有以下格式：

DF2 包含以下内容：

我需要组合两个数据帧，如果 DF2 的“id_no”和“开始日期”与 DF1 匹配，则应在 DF1 中替换它，如果不匹配，则应将其插入到 DF1 中。“id_no”不是唯一的。

预期结果：

我正在循环中从文件夹中读取文件并从中创建数据帧。但是，我收到了这个奇怪的错误TypeError: 'str' object is not callable。请在此处找到代码：

for yr in range (2014,2018):
  cat_bank_yr = sqlCtx.read.csv(cat_bank_path+str(yr)+'_'+h1+'bank.csv000',sep='|',schema=schema)
  cat_bank_yr=cat_bank_yr.withColumn("cat_ledger",trim(lower(col("cat_ledger"))))
  cat_bank_yr=cat_bank_yr.withColumn("category",trim(lower(col("category"))))

代码运行一次迭代，然后停在该行

cat_bank_yr=cat_bank_yr.withColumn("cat_ledger",trim(lower(col("cat_ledger")))) 

出现上述错误。

任何人都可以帮忙吗？

我有以下 PySpark DataFrame：

id   col1   col2
A    2      3
A    2      4
A    4      6
B    1      2

我想堆叠col1并col2获得如下单列：

id   col3
A    2   
A    3
A    4
A    6
B    1
B    2

我怎么能这样做？

df = (
    sc.parallelize([
        (A, 2, 3), (A, 2, 4), (A, 4, 6),
        (B, 1, 2),
    ]).toDF(["id", "col1", "col2"])
)

我在 PySpark 中有以下 DataFrame：

Id      DateActual          DateStart               DateEnd                 SourceCode
107 2019-08-11 00:00:00     null                    null                    1111
107 2019-08-16 00:00:00     2019-08-11 00:00:00     2019-08-18 00:00:00     1111
128 2019-02-11 00:00:00     null                    null                    101
128 2019-02-13 00:00:00     2019-02-11 00:00:00     2019-02-18 00:00:00     168
128 2019-02-14 00:00:00     2019-02-13 00:00:00     2019-02-20 00:00:00     187

我需要替换null值以获得以下结果：

Id      DateActual          DateStart               DateEnd                 SourceCode
107 2019-08-11 00:00:00     2019-08-11 00:00:00     2019-08-18 00:00:00     1111
107 2019-08-16 00:00:00     2019-08-11 00:00:00     2019-08-18 00:00:00     1111
128 2019-02-11 00:00:00     2019-02-11 00:00:00     2019-02-18 00:00:00     101
128 2019-02-13 00:00:00 …

我有一个使用 spark-xml 包转换为数据帧的 XML 文件。数据框具有以下结构：

root
 |-- results: struct (nullable = true)
 |    |-- result: struct (nullable = true)
 |    |    |-- categories: struct (nullable = true)
 |    |    |    |-- category: array (nullable = true)
 |    |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |    |-- value: string (nullable = true)

如果我选择类别列（可能在类别下多次出现）：

df.select((col('results.result.categories.category')).alias("result_categories"))

对于一个记录，结果看起来像

[[result1], [result2]]

我试图压平结果：

[result1, result2]

当我使用 flatten 函数时，我收到一条错误消息：

df.select(flatten(col('results.result.categories.category')).alias("Hits_Category"))
 cannot resolve 'flatten(`results`.`result`.`categories`.`category`)' due to data type mismatch: The argument should …

我有关于年、月和日的三列。如何使用这些在 PySpark 中创建日期？

我有一个如下所示的数据集：

我按年龄分组，平均每个年龄的朋友数量

from pyspark.sql import SparkSession
from pyspark.sql import Row
import pyspark.sql.functions as F

def parseInput(line):
    fields = line.split(',')
    return Row(age = int(fields[2]), numFriends = int(fields[3]))

spark = SparkSession.builder.appName("FriendsByAge").getOrCreate()
lines = spark.sparkContext.textFile("data/fakefriends.csv")
friends = lines.map(parseInput)
friendDataset = spark.createDataFrame(friends)
counts = friendDataset.groupBy("age").count()
total = friendDataset.groupBy("age").sum('numFriends')
res = total.join(counts, "age").withColumn("Friend By Age", (F.col("sum(numFriends)") // F.col("count"))).drop('sum(numFriends)','count')

我得到以下错误：

TypeError: unsupported operand type(s) for //: 'Column' and 'Column'

通常，我在 Python 3.0+ 中使用//并像我在这里预期的那样返回一个整数值，但是，在 PySpark 数据报中， // 不起作用，只有 / 起作用。有什么理由不工作吗？我们必须使用round函数来获取整数值吗？

我有一个 pyspark 数据框：

number  |  matricule      
--------------------------------------------
1       |  ["AZ 1234", "1234", "00100"]                   
--------------------------------------------
23      |  ["1010", "12987"]                   
--------------------------------------------
56      |  ["AZ 98989", "22222", "98989"]                   
--------------------------------------------

在matricule数组中，如果我删除AZ字符串，我会有重复的值。我想删除"AZ"字符串然后删除matricule 数组中的重复值。知道有时我在 之后有一个空格AZ，我也应该将其删除。

我做了一个udf：

def remove_AZ(A)
    for item in A:
        if item.startswith('AZ'):
            item.replace('AZ','')
udf_remove_AZ = F.udf(remove_AZ)
df = df.withColumn("AZ_2", udf_remove_AZ(df.matricule))

我在所有AZ_2列中都为空。

如何从matricule数组中的每个值中删除 AZ然后删除里面的重复项？谢谢

我有一个包含两列的数据框：filename和year。我想filename用year列中的值替换年份值

下表中的第三列显示了要求：

+----------------------------+------+----------------------------+
| filename                   | year | reqd_filename              |
+----------------------------+------+----------------------------+
| blah_2020_v1_blah_blah.csv | 1975 | blah_1975_v1_blah_blah.csv |
+----------------------------+------+----------------------------+
| blah_2019_v1_blah_blah.csv | 1984 | blah_1984_v1_blah_blah.csv |
+----------------------------+------+----------------------------+

代码目前如下所示：

df = df.withColumn('filename', F.regexp_replace(F.col('filename',), '(blah_)(.*)(_v1.*)', <Nothing I put here works>))

简而言之，我想用yeardf 中的列替换第二组

我有一个DataFrame的列ArrayType(StringType)：

+------------------------------------+
|colname                             |
+------------------------------------+
|[foo_XX_foo, bar_YY_bar]            |
|[qwe_ZZ_rty, asd_AA_fgh, zxc_BB_vbn]|
+------------------------------------+

我现在想提取第一个和第二个之间的字符串_，即预期的输出是：

+------------+
|newcolname  |
+------------+
|[XX, YY]    |
|[ZZ, AA, BB]|
+------------+

继这个答案，我试图用expr()用transform，但我没能得到它的工作。即使将所有字符串更改为大写的示例（如上述引用的答案）也不适用于我，但出现以下错误：

pyspark.sql.utils.ParseException：u“ \ nextraneous输入'>'期望{'（'，'SELECT'，...

如何修改中的所有元素ArrayType？我想避免使用udf。