如何在PySpark中跨多个时间间隔使用.filter()操作？
在PySpark中，我们可以使用.filter()操作来筛选DataFrame中的数据。如果需要跨多个时间间隔进行筛选，可以使用Python的datetime模块来处理时间。下面是一个示例代码，演示如何在PySpark中跨多个时间间隔使用.filter()操作：
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from datetime import datetime, timedelta
# 创建SparkSession
spark = SparkSession.builder.getOrCreate()
# 创建示例数据
data = [
("2022-01-01", "A"),
("2022-01-02", "B"),
("2022-01-03", "C"),
("2022-01-04", "D"),
("2022-01-05", "E"),
("2022-01-06", "F")
]
df = spark.createDataFrame(data, ["date", "value"])
# 定义时间间隔
start_date = datetime.strptime("2022-01-02", "%Y-%m-%d")
end_date = datetime.strptime("2022-01-04", "%Y-%m-%d")
# 使用.filter()操作跨多个时间间隔筛选数据
filtered_df = df.filter((col("date") >= start_date) & (col("date") <= end_date))
# 显示筛选后的结果
filtered_df.show()在上面的代码中，我们首先导入了必要的模块和函数，然后创建了一个SparkSession。接下来，我们创建了一个示例数据的DataFrame，其中包含了日期和值两列。
然后，我们使用datetime模块定义了需要筛选的时间间隔，这里我们选择了从2022-01-02到2022-01-04这个时间段。
最后，我们使用.filter()操作结合col()函数来筛选满足条件的数据，条件是日期列大于等于start_date并且小于等于end_date。筛选后的结果存储在filtered_df中，并通过.show()方法进行显示。
请注意，这只是一个示例代码，实际使用时，你需要根据自己的数据和时间间隔进行相应的调整。
关于PySpark和数据处理的更多信息，你可以参考腾讯云的相关产品和文档：
- PySpark开发指南：https://cloud.tencent.com/document/product/849/53223
- 腾讯云数据仓库CDW：https://cloud.tencent.com/product/cdw
- 腾讯云数据湖LakeHouse：https://cloud.tencent.com/product/lakehouse
相关·内容
我正在使用PySpark，几乎要放弃我的问题了。我有两个数据集:一个非常大的(集合A)和一个相当小的(集合B)。它们的形式如下： Data set A:
variable | timestampA
---------------------------------
x | 2015-01-01 09:29:21
y | 2015-01-01 12:01:57
Data set B:
different information | timestampB
-------------------------------------------
info
浏览 10提问于2020-10-08得票数 1
回答已采纳
我是pyspark的新手。我有一个类似下面的文本文件，我想用pyspark解析它，并且只得到一个列表ro，ni，da的花朵缩写。我该怎么做呢？请指教
Plantes缩写
鲜花：
ro rose
尼罗法尔
大丽花
浏览 2提问于2018-04-03得票数 0
我在PySpark中这样做过：
使用DataFrame语句创建一个SELECT，以获取按资产序列号排序的资产数据，然后按时间排序。
使用DataFrame.map()将DataFrame转换为RDD。
使用整理每个资产的所有数据，使用资产的序列号作为密钥。
问题：，我能确定每个资产的数据仍然会在最后一步的RDD中按时间顺序排序吗？
时间顺序对我来说是至关重要的(我需要在每个资产的移动时间窗口上计算统计数据)。当RDD.combineByKey()为给定的密钥组合来自火花集群中不同节点的数据时，该键的数据中是否保留了任何顺序？或者，对于给定的密钥，来自不同节点的数据是否按特定
浏览 1提问于2017-04-26得票数 0
回答已采纳
免责声明:我对pyspark非常陌生，这个问题可能不合适。 我在网上看到了以下代码： # Get the id, age where age = 22 in SQL
spark.sql("select id, age from swimmers where age = 22").show() 现在，我用以下代码尝试使用pyspark进行透视： complete_dataset.createOrReplaceTempView("df")
temp = spark.sql("SELECT core_id from df") 这是我得到的错误： &#
浏览 30提问于2020-06-17得票数 1
我在PySpark中有一个PySpark，它的一个字段有一个嵌套数组值。我想过滤数组包含特定字符串的DataFrame。我不知道该怎么做。
模式如下所示：root |-- name: string (nullable = true) |-- lastName: array (nullable = true) | |-- element: string (containsNull = false)
我希望返回upper(name) == 'JOHN'和lastName列(数组)包含'SMITH'和等式的所有行(就像我对名称所做的那样)。我在列值上找到了
浏览 4提问于2016-06-24得票数 0
回答已采纳
1回答
我的数据集是：
mydata <- data.frame(
x1= as.factor(1:3),
x2= as.factor (4:6),
x3= as.factor(7:9),
x4= as.factor (2:7),
x5= as.factor(1:6),
fx= seq(0,600,len= 600),
time= seq(0,1,len=600)
)%>%
group_by(x1, x2, x3, x4,x5)
我想选择坐标(时间，fx)的点，它与第一个时间点、最后一个时间点、有min(Fx)的点和一半的间隔点相匹配。
浏览 3提问于2016-10-25得票数 0
回答已采纳
1回答
我想存储10个网站的访问日志，每月在DynamoDB上总共收到1000万次访问。在此之后，我想创建一个后端来监视和检测欺诈操作。
DynamoDB能否处理复杂的查询，如：-在指定间隔内具有X跳率的访问者列表--在日期/时间和日期/时间/时间排序和分组之间的流行目标URI
浏览 1提问于2013-08-27得票数 0
回答已采纳
我正在尝试运行中给出的Spark / Python的Logistic回归示例，并且已经成功地使用了Spark1.6和Python2.7。
现在我必须将它移到Spark2.1和Python3.5( 3.6是不兼容的)，我正在使用Ubuntu16.04中的木星笔记本
这段代码工作正常
# Evaluate the model on training data
labelsAndPreds = modelInput.map(lambda p: (p.label, LRmodel.predict(p.features)))
print(labelsAndPreds.count())
print(lab
浏览 2提问于2017-04-12得票数 1
回答已采纳
我有一个pyspark中的dataframe，它有一个时间戳字符串列，格式如下：
"11/21/2018 07:21:49下午“
这是24小时格式的
不管日期如何，我只想根据这个字符串时间戳的时间部分过滤dataframe中的行。例如，我希望保持所有在下午2:00到4:00之间的行都包括在内。
我尝试了以下提取的HH:mm:ss和使用之间的功能，但它是无效的。
# Grabbing only time portion from datetime column
import pyspark.sql.functions as F
time_format = "HH:mm:ss
浏览 3提问于2020-11-23得票数 1
回答已采纳
1回答
我试图从rdd中筛选值为"01-10-2019“的
print("\n ### Remove duplicates in merged RDD:")
insuredata = insuredatamerged_cache.distinct()
print("insuredata: ",type(insuredata))
print("\n ### Increase partition to 8 in merged RDD:")
insuredata.getNumPartitions()
insuredatarepart = insu
浏览 5提问于2020-02-08得票数 1
回答已采纳
我正在使用pyspark将文件加载到下面的表中
LOAD DATA LOCAL INPATH '/user/hive/warehouse/students' OVERWRITE INTO TABLE test_load;
它抱怨说：
pyspark.sql.utils.AnalysisException:当路径字符串在目录结构(如XX/XX/2021-03-02T20:04:27+00:00/file.parquet )中具有时间戳时不存在加载数据输入路径
它与没有时间戳的路径一起工作。怎么解决这个问题？
浏览 1提问于2021-04-23得票数 0
2回答
火花放电数据的慢速滤波
、、、
在过滤熊猫和火星雨数据时，我有一个关于时间差的问题：
import time
import numpy as np
import pandas as pd
from random import shuffle
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df = pd.DataFrame(np.random.randint(1000000, size=400000).reshape(-1, 2))
list_filter = list(range(10000))
shuffl
浏览 1提问于2018-12-12得票数 5
回答已采纳
1回答
我有pyspark数据集，我想按照datetime列将其划分为列车和测试数据集，其中，训练数据集的日期时间应该小于日期时间列的中位数，而测试数据集应该有其余的数据集。
我尝试按datetime列对dataset进行排序，并选择上半部分。但是这只解决了火车部件的问题，我不知道如何从PySpark中的初始数据集中“减除”训练数据集。
train = data.orderBy('datetime').limit(data.count() // 2)
# test = ?
如果PySpark与Pandas ()函数有某种相似之处，那就太好了，但它没有。
浏览 0提问于2019-08-21得票数 0
回答已采纳
1回答
我们的应用程序使用的是旧的框架集。它将布局划分为顶部框架、左侧框架和右侧框架。顶部和左侧框架都有按钮，在右侧框架中我们显示页面。在点击顶部或左侧框架中的任何按钮时，我们首先通过jquery ajax检查会话，然后在右侧框架中显示页面。现在，我们展示了在会话检查过程中，希望在右边的框架中显示一条加载消息。请让我知道我们该怎么做？
浏览 1提问于2014-09-30得票数 0
1回答
我正在尝试基于时间戳[(start1, stop1), (start2, stop2), ...]的元组列表来筛选Pyspark。每个元组表示一个时间窗口。Pyspark数据作为以下结构：
+-------------------+------+
| ts| var|
+-------------------+------+
|2018-09-01 20:10:00| 0|
|2018-09-01 20:12:00| 2|
|2018-09-01 20:13:00| 1|
|2018-09-01 20:17:00| 5|
+-
浏览 1提问于2018-09-02得票数 0
回答已采纳
1回答
我需要找到一种方法来获取所有在pyspark中具有null值的行。
例如，我有以下数据：
+-------+-------+-------+
| c_00| c_01| c_02|
+-------+-------+-------+
1 | null | 0.141| 0.141|
2 | 0.17| 0.17| 0.17|
3 | 0.25| null| 0.25|
4 | 0.135| 0.135| 0.135|
我希望所有列中都有null值的行
+-------+-------+-------+
|
浏览 1提问于2022-01-28得票数 4
回答已采纳
我有一个13M行的pyspark数据帧，我想把它转换成pandas数据帧。然后，将根据其他参数以不同的频率(例如1秒、1分钟、10分钟)对数据帧进行重新采样以进行进一步分析。
从文献[，]中，我发现使用以下任何一行都可以加快pyspark到pandas数据帧之间的转换：
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
spark.conf.set("spark.sql.execution.arrow.enabled", "true")
但是，
浏览 11提问于2021-11-19得票数 0
当测试我们的游戏，这是严重依赖于System.currentTimeMillis()，我们正在经历一个恼人的错误。
我们的游戏使用一组达美时间戳来指示某些事情何时会发生。这些时间戳与正在播放的一段音乐相匹配。
在家里测试一点问题都没有。在我们家里测试的时候，不可能复制这个错误。
但是在汽车里开车的时候，在城市之间进行测试，会让我们在时间戳和音乐之间产生同步问题。我最好的猜测是Android冻结了系统，包括系统定时器，因为它是交换网络，还是寻找信号？
我试着在游戏中插入一个假连接，当我按下某个按钮时，我会让线程休眠几秒钟。这会冻结屏幕(很明显)，但当睡眠结束时，一切仍然同步。
复制这种错误的唯一
浏览 2提问于2013-06-02得票数 4
回答已采纳
我的数据如下。我想按时间戳排序，并使用每个userid的最新样本作为测试数据。我应该如何做训练和测试分离？我尝试的是使用pandas来sort_values时间戳，然后按“userid”分组。但我只得到了一个groupby对象。这样做的正确方法是什么？pyspark是一个更好的工具吗？
得到测试数据的数据帧后，如何拆分数据？显然，我不能使用sklearn的train_test_split。
浏览 3提问于2019-11-13得票数 1
我想当userid=22650984.How在pyspark平台上更新它的价值?谢谢你的帮助。
>>>xxDF.select('userid','registration_time').filter('userid="22650984"').show(truncate=False)
18/04/08 10:57:00 WARN TaskSetManager: Lost task 0.1 in stage 57.0 (TID 874, shopee-hadoop-slave89, executor 9): Task
浏览 8提问于2018-04-08得票数 11
回答已采纳
def filter1(x):
if broadcast_variable1.value[x] > 1:
return False
return True
def filter2(x):
if broadcast_variable2.value[x] < 1:
error_accumulator_variable.add(1)
return False
return True
这些函数包含在我的主要pyspark脚本中。现在，为了便于维护，我想将它们拆分成一个模块文件(我有两个pyspark脚本用于不同的
浏览 1提问于2016-11-10得票数 0
1回答
我有下面的data.frame ( spark )
import findspark
findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
sc = spark.sparkContext
from pyspark.sql import functions as sf
from pyspark.sql.functions import col, when, lit
ddf = spark.createDataFrame([[None, 'Mic
浏览 2提问于2018-05-28得票数 2
回答已采纳
我有一个PySpark代码，它接受多个输入数据，并生成一个数据作为输出。我执行许多火花转换(地图，过滤器，连接，还原键.)但没有火花行动。我执行的唯一操作是在代码的末尾(输出df的count())。代码太长，不能放在这里，但是看起来如下所示：
df1 = HiveContext.sql("select * from db.table1")
df2 = HiveContext.sql("select * from db.table2")
...
...
dfN = HiveContext.sql("select * from db.tableN")
浏览 3提问于2017-07-27得票数 0
回答已采纳
1回答
我在PySpark中观察到一个奇怪的行为。也许你们中的一个会知道发生了什么。如果我这样做：
def create_my_date(mydate):
try:
return mydate.strftime('%Y%m')
except:
return None
df = df.withColumn(
"date_string",
F.udf(create_id, StringType())(df.mydate)
)
df.filter(~df.mydate.isNotNull
浏览 34提问于2018-01-10得票数 0
我正在从SQL server中读取大约900万行的数据，并将其插入到我的datalake中已经存在的表中。 这一过程处理的数据较少，约为100万。 我对sql server使用基本的只读写模式： ? 我的Spark提交如下所示： ? 我的Pyspark配置： ? 我已经尝试重新分区并将内存增加到15，但仍然存在相同的问题。 java.io.IOException:流已损坏 抱歉，我没有访问完整日志的权限
浏览 31提问于2020-07-24得票数 1
Kubernetes名称空间到底是什么?它与linux名称空间(mnt、pid、net、ipc、uts、user、cgroup)有什么不同？
浏览 20提问于2020-05-06得票数 3
给定如下所示的DataFrame，是否可以在保持模式完整的同时过滤出PySpark中列集合的一些键(MapType(StringType，StringType，True))？
root
|-- id: string (nullable = true)
|-- collection: map (nullable = true)
| |-- key: string
| |-- value: string
浏览 3提问于2017-06-22得票数 4
回答已采纳
我正在使用codahale指标(现在是dropwizard指标)来监控我系统中发生的一些“事件”。我使用counters指标来跟踪“事件”发生的次数。
我检查了记者为我的计数器指标打印的值，发现该值似乎一直在增加(并且从未下降)。这似乎合乎逻辑，因为每当我的“事件”发生时，我总是使用metrics.inc()函数。
What I really want is to get count of my 'event' happening between two reporting times，为此，我需要在每次报告指标时重置计数器，但我在计数器指标中找不到任何选项来执行此操作。cod
浏览 4提问于2016-07-01得票数 10
我想对排序后的Pyspark数据帧进行基于时间的训练测试拆分。假设前300行将在训练集中，下200行将在测试拆分中。 我可以用以下命令选择前300行： train = df.show(300) 但是如何从Pyspark dataframe中选择最后200行呢？
浏览 14提问于2019-03-13得票数 1
我将我的一段遗留代码从一个列表上的简单for循环替换为Java 8流和过滤器。
我的for循环如下所示：
List<Integer> numbers = Arrays.asList(1,2,3,4,5,6,7,8,9,10);
for(int i: numbers) {
if((i > 4) && (i % 2 == 0)) {
System.out.println("First even number in list more than 4 : " + numbers.get(i))
break;
浏览 5提问于2021-05-12得票数 0
回答已采纳
2回答
我正在用PySpark处理csv文件的PCA。我得到了一些奇怪的行为；我的代码有时运行良好，但有时返回此错误：
File "C:/spark/spark-2.1.0-bin-hadoop2.7/bin/pca_final2.py", line 25, in <module>
columns = (fileObj.first()).split(';')
File "C:\spark\spark-2.1.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\rdd.py", line 1361,
浏览 1提问于2017-03-14得票数 4
回答已采纳
我在eclipse上为android编写了一个应用程序(订单信息、订单细节、使用webservice和webdav预订时间戳)。
我的代码是否可用于其他平台(如ios、windows )？
谢谢你对我的照顾。迈克尔
浏览 2提问于2013-12-13得票数 1
回答已采纳
我在Cassandra中将时间戳存储为YYYY-mm-dd HH:MM:SSZ，并且我能够在cql shell中过滤数据以获得特定的时间范围，但是当我在pyspark数据帧上尝试同样的操作时，我在过滤后的数据帧中没有得到任何值。
有谁能帮我在pyspark中找到合适的datetime格式吗？
谢谢。
浏览 6提问于2020-03-05得票数 1
我有以下映射的超类：
@FilterDef(
name = "byCompanyUuid",
parameters = @ParamDef(
name = "companyUuid",
type = "string"
)
)
@Filter(
name = "byCompanyUuid",
condition = "COMPANY.UUID = :companyUuid&#
浏览 0提问于2020-12-18得票数 0
1回答
规范一列dataframe
、、、、
我试图使用L1库对数据框架中列的值进行pyspark ML规范化。以下是我的密码。但它做不到。你能帮我弄清楚这段代码有什么问题吗？
from pyspark.ml.feature import Normalizer
y = range(1,10)
data = spark.createDataFrame([[float(e), ] for e in y])
#data.select('_1').show()
normalizer = Normalizer(p=1.0, inputCol="_1", outputCol="features")
浏览 0提问于2017-06-09得票数 1
回答已采纳
1回答
我把下面的UDF从派克的网站，因为我试图了解，如果有一个性能改善。我做了大量的数字，但两者的时间几乎相同，我做错了什么？
谢谢!
import pandas as pd
from pyspark.sql.functions import col, udf
from pyspark.sql.types import LongType
import time
start = time.time()
# Declare the function and create the UDF
def multiply_func(a, b):
return a * b
multiply = udf(
浏览 1提问于2020-05-12得票数 3
回答已采纳
1回答
火花源生成日期序列
、、、
我正在尝试生成一个日期序列
from pyspark.sql import functions as F
df1 = df.withColumn("start_dt", F.to_date(F.col("start_date"), "yyyy-mm-dd")) \
.withColumn("end_dt", F.to_date(F.col("end_date"), "yyyy-mm-dd"))
df1.select("start_dt", "end_dt&
浏览 2提问于2020-03-17得票数 3
我是PySpark的初学者，最近我尝试将一个简单的python应用程序(批量调整图片)提交到我的星星团中。我可以通过py魅力成功地运行应用程序，当我提交我的应用程序火花时，图像也会被调整大小。
这是我最初的Python代码：
import os
from PIL import Image
size_64 = (64,64)
for f in os.listdir('.')
if f.endswith('.jpg'):
i = Image.open(f)
fn, fext = os.path.splitext(f)
浏览 5提问于2016-10-28得票数 3
回答已采纳
1回答
我需要用'modifiedDate‘列在Pyspark中创建delta表。行更新时，修改time的值应更改为当前时间。我曾想过要做一个触发器，但却找不到任何关于在带有delta表的Pyspark中使用触发器的信息。我该怎么做？
浏览 1提问于2022-01-30得票数 1
回答已采纳
1回答
我有一个名为h2_df、列为"parsed_date“(dtype: date)和"id”(dtype: bigint)的pyspark dataframe，如下所示： +-------+-----------+
| id|parsed_date|
+-------+-----------+
|1471783| 2017-12-18|
|1471885| 2017-12-18|
|1472928| 2017-12-19|
|1476917| 2017-12-21|
|1477469| 2017-12-22|
|1478190| 2017-12-22|
|1478570
浏览 7提问于2021-01-25得票数 0
回答已采纳
我想要创建SuiteScript2.0，它将从保存的搜索中搜索和获取数据，但是我想安排这个脚本每天在特定的时间运行。
在套件脚本(ScheduledScript)中是否有一种在特定间隔上执行脚本的方法？
浏览 0提问于2019-02-14得票数 1
1回答
列中的搜索值
、、
我想搜索列是否包含值。 import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.types import *
import pandas as pd
df_init = pd.DataFrame({'id':['1', '2'], 'val':[100, 200]})
spark = SparkSession.builder.appName('pandasToSparkDF').getOrCreate()
mySchema
浏览 16提问于2020-09-29得票数 1
回答已采纳
2回答
要Py的SQL查询(Spark)
、、、、
我有以下SQL查询，我想将其转换为pyspark。我想使用两个列pp和gender，并在pyspark中执行以下操作 %sql
SELECT pp
, SUM(CASE WHEN Gender = 'M'
THEN 1.0 ELSE 0.0 END) /
COUNT(1) AS gender_score
, count(1) AS total
FROM df
WHERE gender in ('M', 'F')
GROUP BY pp
HAVING
浏览 12提问于2020-12-16得票数 0
1回答
考虑到同样的情况，微软在他们的蓝色搜索示例中使用了酒店搜索系统，但是蔚蓝搜索应该为开放的酒店提供过滤器。
如何存储有关开放时间的信息(源模型包含每一天的两个间隔--午餐前/午餐后)，以便对过滤器有用？
我找到的最简单的解决方案(肯定不是最漂亮的解决方案)是为每一天创建字段，包括时间间隔以及开始和停止时间戳(只输入时间)。然后在过滤器参数中使用适当的日值和当前时间。整个解决方案需要创建7x2x2字段。
谁有更好的解决方案，也许使用默认功能的Azure搜索引擎？
浏览 3提问于2015-09-02得票数 0
回答已采纳
2回答
我正在尝试使用Phoenix连接器从PySpark读取和写入HBase。
我已经看到中的示例代码复制了此处的示例代码，以便于参考：
df.write \
.format("org.apache.phoenix.spark") \
.mode("overwrite") \
.option("table", "TABLE1") \
.option("zkUrl", "localhost:2181") \
.save()
我已经能够从pyspark读写到hbase了。然而，这个例子只给
浏览 5提问于2018-01-06得票数 0
1回答
如何使从卡桑德拉流到星火？
、、、、
我有下面的星火作业
from __future__ import print_function
import os
import sys
import time
from random import random
from operator import add
from pyspark.streaming import StreamingContext
from pyspark import SparkContext,SparkConf
from pyspark.streaming.kafka import KafkaUtils
from pyspark.sql import SQLCon
浏览 1提问于2016-01-26得票数 7
回答已采纳
我是星火公司的新手。我已经安装了Spark，可以成功地运行以下测试代码：
import findspark
findspark.init()
import pyspark # only run after findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df = spark.sql('''select 'spark' as hello ''')
df.show()
但是我在pyspa
浏览 2提问于2022-11-20得票数 1
1回答
如何对星火数据进行逐行并行计算？
、、、、
我收集了30万点，我想计算它们之间的距离。
id x y
0 0 1 0
1 1 28 76
…
因此，我在这些点之间做了一个笛卡儿积，然后我过滤，因为我只保留一个点的组合。实际上，就我的目的而言，(0, 1)与(1,0)之间的距离是相同的。
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
from pyspark.sql.functions import udf
from pyspark.sql.types import Intege
浏览 2提问于2021-02-04得票数 2
回答已采纳
如何从这篇文章中扩展下面的函数来计算'c‘和'd’的中位数：Calculate a grouped median in pyspark # udf for median
import numpy as np
import pyspark.sql.functions as func
def median(values_list):
med = np.median(values_list)
return float(med)
udf_median = func.udf(median, FloatType())
group_df = df.groupby([&#
浏览 26提问于2020-09-29得票数 0
回答已采纳
1回答
我的日期时间字段格式是: 2016-10-15 00:00:00在使用推断模式将我的数据保存到拼图文件时，我有一些行不符合此格式。
如何在PySpark中集中删除它们？
它在我的UDF中引起了问题。
浏览 0提问于2017-01-02得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
