相关疑难解决方法(0)

我见过各种各样的人都认为这Dataframe.explode是一种有用的方法,但它会导致比原始数据帧更多的行,这根本不是我想要的.我只是想做Dataframe相当于非常简单:

rdd.map(lambda row: row + [row.my_str_col.split('-')])

它看起来像:

col1 | my_str_col
-----+-----------
  18 |  856-yygrm
 201 |  777-psgdg

并将其转换为:

col1 | my_str_col | _col3 | _col4
-----+------------+-------+------
  18 |  856-yygrm |   856 | yygrm
 201 |  777-psgdg |   777 | psgdg

我知道pyspark.sql.functions.split(),但它导致嵌套数组列而不是我想要的两个顶级列.

理想情况下,我希望这些新列也可以命名.

Spark现在提供可在数据帧中使用的预定义函数,并且它们似乎已经过高度优化.我最初的问题是更快,但我自己做了一些测试,发现至少在一个实例中,spark函数的速度提高了大约10倍.有谁知道为什么会这样,什么时候udf会更快(仅适用于存在相同spark函数的情况)？

这是我的测试代码(在Databricks社区上运行):

# UDF vs Spark function
from faker import Factory
from pyspark.sql.functions import lit, concat
fake = Factory.create()
fake.seed(4321)

# Each entry consists of last_name, first_name, ssn, job, and age (at least 1)
from pyspark.sql import Row
def fake_entry():
  name = fake.name().split()
  return (name[1], name[0], fake.ssn(), fake.job(), abs(2016 - fake.date_time().year) + 1)

# Create a helper function to call a function repeatedly
def repeat(times, func, *args, **kwargs):
    for _ in xrange(times):
        yield func(*args, **kwargs)
data = list(repeat(500000, fake_entry))
print …

我有一个DF,我有bookingDt和arrivalDt列.我需要找到这两个日期之间的所有日期.

示例代码:

df = spark.sparkContext.parallelize(
            [Row(vyge_id=1000, bookingDt='2018-01-01', arrivalDt='2018-01-05')]).toDF()
diffDaysDF = df.withColumn("diffDays", datediff('arrivalDt', 'bookingDt'))
diffDaysDF.show()

代码输出:

+----------+----------+-------+--------+
| arrivalDt| bookingDt|vyge_id|diffDays|
+----------+----------+-------+--------+
|2018-01-05|2018-01-01|   1000|       4|
+----------+----------+-------+--------+

我尝试的是找到两个日期之间的天数,并使用timedelta函数计算所有日期explode.

dateList = [str(bookingDt + timedelta(i)) for i in range(diffDays)]

预期产量:

基本上,我需要建立一个DF与对之间的每个日的记录bookingDt和arrivalDt,包容性.

+----------+----------+-------+----------+
| arrivalDt| bookingDt|vyge_id|txnDt     |
+----------+----------+-------+----------+
|2018-01-05|2018-01-01|   1000|2018-01-01|
+----------+----------+-------+----------+
|2018-01-05|2018-01-01|   1000|2018-01-02|
+----------+----------+-------+----------+
|2018-01-05|2018-01-01|   1000|2018-01-03|
+----------+----------+-------+----------+
|2018-01-05|2018-01-01|   1000|2018-01-04|
+----------+----------+-------+----------+
|2018-01-05|2018-01-01|   1000|2018-01-05|
+----------+----------+-------+----------+

我正在使用PySpark处理一个庞大的数据集,我希望根据另一个数据框中的字符串过滤数据帧.例如,

dd = spark.createDataFrame(["something.google.com","something.google.com.somethingelse.ac.uk","something.good.com.cy", "something.good.com.cy.mal.org"], StringType()).toDF('domains')
+----------------------------------------+
|domains                                 |
+----------------------------------------+
|something.google.com                    |
|something.google.com.somethingelse.ac.uk|
|something.good.com.cy                   |
|something.good.com.cy.mal.org           |
+----------------------------------------+  

dd1 =  spark.createDataFrame(["google.com", "good.com.cy"], StringType()).toDF('gooddomains')
+-----------+
|gooddomains|
+-----------+
|google.com |
|good.com.cy|
+-----------+

我认为domains和gooddomains有效的域名.

我想要做的是过滤掉dd不结束的匹配字符串dd1.所以在上面的例子中,我想过滤掉第1行和第3行,最后得到

+----------------------------------------+
|domains                                 |
+----------------------------------------+
|something.google.com.somethingelse.ac.uk|
|something.good.com.cy.mal.org           |
+----------------------------------------+  

我目前的解决方案(如下所示)只能考虑最多3个字的域名.如果我要添加say,verygood.co.ac.ukin dd1(即白名单),那么它将失败.

def split_filter(x, whitelist):
    splitted1 = x.select(F.split(x['domains'], '\.').alias('splitted_domains'))
    last_two = splitted1.select(F.concat(splitted1.splitted_domains[F.size(splitted1.splitted_domains)-2], \
       F.lit('.'), \
       splitted1.splitted_domains[F.size(splitted1.splitted_domains)-1]).alias('last_two'))
    last_three = splitted1.select(F.concat(splitted1.splitted_domains[F.size(splitted1.splitted_domains)-3], \
       F.lit('.'), \
       splitted1.splitted_domains[F.size(splitted1.splitted_domains)-2], \
       F.lit('.'), \
       splitted1.splitted_domains[F.size(splitted1.splitted_domains)-1]).alias('last_three'))
    x = …

我正在尝试按以下方式过滤我的 pyspark 数据框：我有一列long_text包含数字，一列包含数字。如果长文本包含number我想保留该列。我正在尝试使用 SQLLIKE语句，但似乎无法将其应用于另一列（此处number）我的代码如下：

from pyspark.sql.functions import regexp_extract, col, concat, lit
from pyspark.sql.types import *
PN_in_NC = (df
        .filter(df.long_text.like(concat(lit("%"), df.number, lit("%"))))))

我收到以下错误： Method like([class org.apache.spark.sql.Column]) does not exist。

我尝试了多种方法来修复它（例如'%number%'在过滤器之前将字符串创建为列，而不是使用lit, using '%' + number + '%'）但没有任何效果。如果LIKE不能应用于另一列，是否有其他方法可以做到这一点？

我有一个这样的数据框，如果列n大于一个，我想将行重复n次：

A   B   n  
1   2   1  
2   9   1  
3   8   2    
4   1   1    
5   3   3 

然后像这样转换：

A   B   n  
1   2   1  
2   9   1  
3   8   2
3   8   2       
4   1   1    
5   3   3 
5   3   3 
5   3   3 

我想我应该使用explode，但我不明白它的工作原理...
谢谢

我正在使用 pyspark 2.1，并且我有一个数据框，其中有两列，日期格式如下：

Column A ,  START_DT       ,  END_DT
1        ,  2016-01-01     ,  2020-02-04
16       ,  2017-02-23     ,  2017-12-24

我想过滤和2018-12-31之间的某个日期（例如 ）（在该示例中，将过滤第二行）。START_DTEND_DT

START_DT和列都END_DT已经是日期格式，我正在寻找像 sql 这样的方法：

SELECT *
FROM  MYTABLE  WHERE  '2018-12-31' BETWEEN start_dt AND end_dt

我有一个 spark 数据框，其中包含一段时间内某些商店中某些产品的销售预测数据。如何计算下 N 个值的窗口大小的预测的滚动总和？

输入数据

+-----------+---------+------------+------------+---+
| ProductId | StoreId |    Date    | Prediction | N |
+-----------+---------+------------+------------+---+
|         1 |     100 | 2019-07-01 | 0.92       | 2 |
|         1 |     100 | 2019-07-02 | 0.62       | 2 |
|         1 |     100 | 2019-07-03 | 0.89       | 2 |
|         1 |     100 | 2019-07-04 | 0.57       | 2 |
|         2 |     200 | 2019-07-01 | 1.39       | 3 |
|         2 |     200 | 2019-07-02 …

嗨，我有 2 列的数据框：

+----------------------------------------+----------+
|                  Text                  | Key_word |
+----------------------------------------+----------+
| First random text tree cheese cat      | tree     |
| Second random text apple pie three     | text     |
| Third random text burger food brain    | brain    |
| Fourth random text nothing thing chips | random   |
+----------------------------------------+----------+

我想生成第 3 列，其中一个单词出现在文本中的 key_word 之前。

+----------------------------------------+----------+-------------------+--+
|                  Text                  | Key_word | word_bef_key_word |  |
+----------------------------------------+----------+-------------------+--+
| First random text tree cheese cat      | tree     | text              |  |
| …

我目前正在尝试找出如何通过列参数将String-format参数传递给to_date pyspark函数。

具体来说，我有以下设置：

sc = SparkContext.getOrCreate()
df = sc.parallelize([('a','2018-01-01','yyyy-MM-dd'),
                      ('b','2018-02-02','yyyy-MM-dd'),
                      ('c','02-02-2018','dd-MM-yyyy')]).toDF(
                    ["col_name","value","format"])

我当前正在尝试添加一个新列，其中将F.col（“ value”）列中的每个日期（它是一个字符串值）解析为一个日期。

对于每种格式，可以分别使用

df = df.withColumn("test1",F.to_date(F.col("value"),"yyyy-MM-dd")).\
        withColumn("test2",F.to_date(F.col("value"),"dd-MM-yyyy"))

但是，这给了我2个新列-但我希望有1个列包含两个结果-但使用to_date函数似乎无法调用该列：

df = df.withColumn("test3",F.to_date(F.col("value"),F.col("format")))

此处将引发错误“列对象不可调用”。

是否可以对所有可能的格式采用通用方法（这样我就不必为每种格式手动添加新列）？