PySpark:获取最近3天的数据

文章/答案/技术大牛

发布

1回答

pyspark

给定一个包含日期的数据帧，我希望只获取数据帧中最近3天的可用行 |id|date||2|2019-11-30||4|2019-11-28F.desc('date')).distinct().limit(3).collect() df.filter(df.date.isin(F.lit(dates))).show(10,False) 我得到的错误是

浏览 40提问于2020-08-26得票数 1

回答已采纳

1回答

socket.timeout mongoDB火花放电

python、mongodb、timeout、pyspark、bigdata

python文件执行一个查询，从mongoDB获取一些数据，然后他们使用星火中的映射操作来处理这些数据。回溯(最近一次调用)：文件"/home/ana/computational_tools_for_big_data/project/review_analysis.py"，

浏览 4提问于2016-11-27得票数 4

2回答

验证PySpark数据格式的日期列

apache-spark、date、validation、pyspark、typeerror

我想验证PySpark数据的日期列。我知道如何为熊猫做这件事，但不能让它对PySpark有效。ValueError:它适用于熊猫数据采集但我不能让它为PySpark工作。获取以下错误： sparkDF = sparkDF['DOB'].apply(lambda x:

浏览 8提问于2022-10-18得票数 1

回答已采纳

2回答

Pyspark读取csv

python、apache-spark、pyspark

新的pyspark，并希望读取csv文件到数据帧。看起来不能读。有什么帮助吗？from pyspark.sql import SQLContextfrom pyspark.sql import Row rdd.cou

浏览 3提问于2018-05-14得票数 1

回答已采纳

1回答

在PySpark中添加具有滚动最新优先级的列

python、pyspark

我有一个包含客户、日期和交易类型列表的pyspark dataframe。X1 || B | 7 | X1 || B | 7 | X1 | X1 | +----------+-----+------+-------------+ 因此，对于X类型，它只获取当前行中的类型，但对于Y

浏览 35提问于2019-06-11得票数 0

回答已采纳

1回答

有没有人能够在Spark中使用elasticsearch xpack sql？

apache-spark、elasticsearch、pyspark、apache-spark-sql

使用PySpark，我试图从elasticsearch中读取数据。: { } 但是，最近我在kibana上尝试了_xpack/sql，在其他SQL客户端上尝试了JDBC，它们在获取数据方面都工作得很好。但是，当我尝试在我的pyspark代码中引用_xpack时，我得到了以下错误： Py4JJavaError

浏览 0提问于2019-01-31得票数 3

1回答

使用配置单元元数据读取HDFS文件- Pyspark

python、dataframe、pyspark、apache-spark-sql、pyspark-sql

我是PySpark新手，正在尝试读取HDFS文件(上面创建了hive表)并创建PySpark数据帧。通过PySpark读取配置单元表非常耗时。有没有什么方法可以动态获取hive列名(用作dataframe中的模式)？我希望将文件位置、表名和数据库名作为输入传递给aa程序/函数，以便从配置单元元数据(可能是元数据xml)中获取模式/列名称，并作为dataframe返回。请指教

浏览 15提问于2019-03-06得票数 0

回答已采纳

1回答

运行spark.read.json时在json中找到重复列，即使没有重复列

azure、pyspark、apache-spark-sql、azure-synapse

在PySpark和Synapse数据流中，我遇到了非常奇怪的错误。 AnalysisException:在数据模式中找到重复列：amendationcommentkey、amendationreasonkey、amendationregulatoryproofkey回溯(最

浏览 3提问于2021-11-25得票数 0

1回答

ModuleNotFoundError:没有名为'sklearn‘的模块

python、python-3.x、pyspark、scikit-learn、user-defined-functions

我正在运行一段代码，它使用pickled来使用一个泡沫化的sklearn模型来获取数据。在我将引擎更改为Python3.6之前，这在Python2.7中似乎没有问题。每当我尝试将dataframe写入数据库时，或者在另一个实例中，当我尝试将数据写入数据库时，我都会得到ModuleNotFoundError: No module named 'sklearn'。：org.apache.spark.SparkException:由于阶段失败而中止作业:阶段1.0中

浏览 12提问于2022-02-09得票数 0

1回答

在3.6之前不支持python 2和3

python、apache-spark、pyspark

我的代码编译成功，并通过了所有测试用例，但由于推荐错误而陷入困境：我现在该怎么做？

浏览 4提问于2021-12-25得票数 -1

1回答

在Pyspark/Hive中处理更改的数据类型

python、apache-spark、pyspark、apache-spark-sql

在解析pyspark中不一致的数据类型时，我遇到了一个问题。如下面的示例文件所示，SA键总是包含一个字典，但有时它可以显示为string值。当我尝试获取列SA.SM.Name时，会得到如下所示的异常。对于SA.SM.Name列，如何将null设置为除了JSONs之外的值的pyspark/hive。有人能帮帮我吗？我试图转换成不同的数据类型，但是没有工作，或者我可能做错了什么。spark.read.json

浏览 6提问于2020-10-10得票数 0

1回答

ValueError:时间数据“00.00.00”火花

python、datetime、pyspark、apache-spark-sql、azure-databricks

我正在从"mm.dd.yy“格式的中获取日期值。我想把它转换成"mm.dd.yyyy“格式。from pyspark.sql.functions import udf def change_date：‘SparkException:由于阶段失败而中止的作业:阶段302.0中的任务23失败了4次，

浏览 2提问于2020-04-16得票数 1

回答已采纳

1回答

使用PySpark从Blob存储容器加载CSV文件

csv、apache-spark、pyspark、azure-blob-storage

我无法使用Jupyter Notebook中的PySpark将CSV文件从Azure Blob Storage直接加载到RDD中。我已经通读了几乎所有其他类似问题的答案，但我还没有找到关于我想要做什么的具体说明。我知道我也可以使用Pandas将数据加载到Notebook中，但之后我需要将Panda DF转换为RDD。我理想的解决方案应该是这样的，但是这个特定的代码给我的错误是它不能推断出CSV的模式。inferSch

浏览 21提问于2019-04-28得票数 0

1回答

pyspark kafka流偏移量

apache-spark、pyspark、apache-kafka、streaming、offset

我从下面的链接中获得了关于kafka主题偏移流的pyspark。from pyspark.streaming import StreamingContextfrom pyspark.streaming.kafkaKafkaUtils.createDirectStream(stream, [topic], kafkaParams, 参考链接：我不知道在下面提

浏览 0提问于2018-10-06得票数 0

3回答

在Spark DataFrame中将空值转换为空数组

apache-spark、dataframe、apache-spark-sql、apache-spark-1.5

我有一个Spark数据帧，其中一列是整数数组。该列可以为空，因为它来自左外部联接。我希望将所有null值转换为一个空数组，这样以后就不必处理null值了。有没有其他简单的方法来转换空值？如果是相关的，下面是本专栏的模式：| |-- element: integer (containsNull = false)

浏览 5提问于2016-01-08得票数 24

回答已采纳

2回答

如何在Palantir Foundry中的PySpark架构上创建空数据集？

palantir-foundry、foundry-code-repositories

我有一个PySpark模式，它描述数据集的列及其类型(我可以手工编写这些数据集，也可以从现有的数据集获取，然后转到“PySpark”选项卡，然后“复制PySpark模式”)。我想要一个包含此架构的空数据集，例如，它可以用作只写回写的本体对象的支持数据集。我怎样才能在铸造中创造这个呢？

浏览 16提问于2022-08-18得票数 3

回答已采纳

3回答

空值，同时将字符串转换为使用pyspark的datetime。

apache-spark、pyspark

最近，我开始在databricks上使用pyspark，在将下面的字符串转换为DateTime数据类型时，如何获得空值是很有趣的。我在这里看过几篇关于如何做到这一点的文章，但似乎没有一篇对我有用。样本数据 001 34 12/1/2010 8:26 004

浏览 14提问于2021-07-15得票数 1

回答已采纳

1回答

火花中的dataframe.show()和dataframe.take()有什么区别？为了提高业绩，我们需要增加什么？

apache-spark、apache-spark-sql

我正在使用dataframes从parquet文件读取数据，并创建一个临时视图，并在临时视图之上运行SQL查询。filename.parquet").createOrReplaceTempView("temptable")要检查df的结果，我使用的是df.show()，但是执行需要更多的时间，如果使用df.take(10)，我没有看到任何区别。为了更好地检查结果，应该使用take(

浏览 1提问于2018-12-21得票数 1

2回答

局部敏感散列的火花实现

apache-spark、locality-sensitive-hash

作为一个项目的一部分，我正在为我的研究，我正在寻找一种方法使用散列功能的LSH与火花。有没有办法这样做？

浏览 1提问于2014-12-31得票数 2

回答已采纳

1回答

优化迭代数据帧的复杂循环的最佳方法

python、pandas、dataframe、pyspark、luigi

我这里有几个方法，它们花费的时间比我想要的要长。我现在遇到了困难，因为我没有看到任何明显的方法来以更有效的方式编写这些方法。作为后台，代码所做的是处理一个销售数据集，以便找到与同一客户相关的以前的销售订单。但是，正如您将看到的，中间有很多业务逻辑，这可能会减慢速度。我正在考虑将其重构为PySpark工作，但在我这样做之前，我想知道这是不是完成这项工作的最好方法。我将非常感谢这里的

浏览 18提问于2021-11-02得票数 0

回答已采纳

点击加载更多