如何使用Spark Dstream进行简单的随机采样？(使用spark 1.6.1的pyspark)
Spark Dstream是Spark Streaming的核心抽象，用于处理实时数据流。在使用Spark Dstream进行简单的随机采样时，可以按照以下步骤进行操作：
- 导入必要的库和模块：from pyspark import SparkContext from pyspark.streaming import StreamingContext
- 创建SparkContext和StreamingContext对象：sc = SparkContext(appName="RandomSampling")
ssc = StreamingContext(sc, batchDuration)其中，
appName是应用程序的名称，batchDuration是批处理的时间间隔，单位为秒。 - 创建Dstream对象：dstream = ssc.socketTextStream(hostname, port)其中，
hostname是数据源的主机名，port是数据源的端口号。 - 进行随机采样：sampled_dstream = dstream.sample(withReplacement, fraction)其中，
withReplacement是一个布尔值，表示是否进行有放回采样，fraction是采样的比例。 - 打印采样结果：sampled_dstream.pprint()
- 启动StreamingContext并等待程序终止：ssc.start() ssc.awaitTermination()
完整的代码示例如下：
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
# 创建SparkContext和StreamingContext对象
sc = SparkContext(appName="RandomSampling")
ssc = StreamingContext(sc, batchDuration)
# 创建Dstream对象
dstream = ssc.socketTextStream(hostname, port)
# 进行随机采样
sampled_dstream = dstream.sample(withReplacement, fraction)
# 打印采样结果
sampled_dstream.pprint()
# 启动StreamingContext并等待程序终止
ssc.start()
ssc.awaitTermination()对于上述代码中的参数，可以根据实际需求进行调整。此外，腾讯云提供了Spark相关的产品和服务，如腾讯云EMR（Elastic MapReduce），可用于大数据处理和分析。具体产品介绍和文档可以参考腾讯云EMR的官方网站：腾讯云EMR。
相关·内容
我想从Kafka主题中读取使用Python的Spark streaming的值。我使用的是带有spark-streaming-kafka-0-8支持的DStream应用程序接口(尽管已被弃用)。我的代码如下： from pyspark import SparkContext, SparkConf
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
# Local SparkContext and StreamingContext (batch interv
浏览 22提问于2020-04-30得票数 0
我的spark streaming作业(spark 1.6.1，kafka 0.9.0)正在消耗20个分区的Kafka主题。在oracle DB中维护偏移量。
在作业启动时，我会从oracle读取偏移量(只读一次)，并在处理后将偏移量写入oracle。
我的作业成功运行了8个小时，然后失败了，原因如下。在故障期间，kafka主题、spark程序、oracle代码没有任何变化。
谁能告诉我为什么我在一个正在运行的spark流作业中得到这个错误？
16/11/02 08:09:21 ERROR JobScheduler: Error generating jobs for time 1478074
浏览 3提问于2016-11-03得票数 4
1回答
火花流
、
我编写了这段代码，用于吡火花上的虹膜分类，但是我得到了一个错误"'RDD‘对象没有属性'_jdf’“。我已经将RDD更改为dataframe，但它告诉我们，"RDD是不可移植的“。请帮我解决！非常感谢。
这里是我的代码：
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.ml import PipelineModel, Pipeline
from pyspark.sql import Row, DataFrame
from py
浏览 0提问于2018-08-17得票数 2
我想用Kafka和Spark做情感分析。我想做的是从Kafka中读取流数据，然后使用Spark对数据进行批处理。在此之后，我想使用我使用Tensorflow创建的函数sentimentPredict()来分析批处理。这就是我到目前为止所做的。
import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2 pyspark-shell'
# Spark
from pyspark import S
浏览 0提问于2018-04-09得票数 8
我正在尝试运行一个流应用程序来计算特定用户的tweet。生产者守则：
# -*- coding: utf-8 -*-
import tweepy
import json
import base64
from kafka import KafkaProducer
import kafka.errors
# Twitter API credentials
CONSUMER_KEY = "***"
CONSUMER_SECRET = "***"
ACCESS_TOKEN = "***"
ACCESS_TOKEN_SECRET = &#
浏览 3提问于2021-12-06得票数 0
回答已采纳
我正在使用pyspark和elasticsearch (py库)，在更新ES中的一个文档时，我得到了以下错误。
2021-09-08 06:31:49 ERROR JobScheduler:91 - Error running job streaming job 1631082700000 ms.1
org.apache.spark.SparkException: An exception was raised by Python:
Traceback (most recent call last):
File "/usr/bin/spark-2.4.0-bin-hadoop2.7
浏览 4提问于2021-09-08得票数 0
回答已采纳
我正在探索通过PySpark进行流传输的火花，并在尝试将transform函数与take一起使用时出错。
我可以通过sortBy和DStream的结果来成功地对DStream使用transform和pprint。
author_counts_sorted_dstream = author_counts_dstream.transform\
(lambda foo:foo\
.sortBy(lambda x:x[0].lower())\
.sortBy(lambda x:x[1],ascending=False))
author_counts_sorted_dstream.ppr
浏览 2提问于2017-01-05得票数 4
回答已采纳
1回答
我们有一个MapR集群，它在上面运行，但现在它突然停止了，甚至不能在mapr演示集群上工作。我们运行的是MapR 5.1和Spark 1.6.1。
from pyspark import SparkConf, SparkContext
from pyspark import HiveContext
from pyspark.sql import DataFrameWriter
conf = SparkConf().setAppName('test')
sc = SparkContext(conf=conf)
sqlContext = HiveContext(sc)
df =
浏览 1提问于2016-09-29得票数 2
我下载了anaconda3，spark1.6.1，java，git等等.
我建立了这样的环境
$ nano .bashrc
export PATH="/home/moon/anaconda3/bin:$PATH"
export SCALA_HOME=/usr/local/src/scala/scala-2.12.1
export PATH=$SCALA_HOME/bin:$PATH
export SPARK_PATH=~/spark-1.6.1-bin-hadoop2.6
export PATH=$SPARK_PATH/bin:$PATH
export ANACO
浏览 2提问于2017-03-17得票数 1
我有一个简短的脚本在python中，我需要得到异常，但只有一个valueError而不是完整的内容。我用代码来解释：
try:
r = str(ML_engine.Create_ML_Alg_Python(sc, m))
ML_engine.updateModel('success',r,m)
return r
except Exception as inst:
ML_engine.updateModel(str(inst), -200, m)
return str(inst)
异常发生时，在python控制台视图中：
File "/hom
浏览 0提问于2016-06-18得票数 1
为了编写一个独立的脚本，我想直接从Python启动和配置一个Spark上下文。使用PySpark的脚本，我可以通过以下方式设置驱动程序的内存大小：
$ /opt/spark-1.6.1/bin/pyspark
... INFO MemoryStore: MemoryStore started with capacity 511.5 MB ...
$ /opt/spark-1.6.1/bin/pyspark --conf spark.driver.memory=10g
... INFO MemoryStore: MemoryStore started with capacity 7.0 GB .
浏览 6提问于2016-06-23得票数 7
回答已采纳
我正在将数据从netezza加载到一个dataframe中，然后尝试写到dashdb。我正在使用ibmdpy尝试将数据加载到bluemix上的dashdb中。Ibmdpy需要一个熊猫数据，所以我把星火数据转换成熊猫，然后加载到dashdb中。
all_disputes_df = sqlContext.read.format('jdbc').options(url='jdbc:netezza://pda1-wall.pok.ibm.com:5480/BACC_PRD_ISCNZ_GAPNZ', user=user, password=password, dbtab
浏览 2提问于2016-06-14得票数 1
2回答
我是spark & pyspark的新手。
我正在将一个小的csv文件(~40k)读取到数据帧中。
from pyspark.sql import functions as F
df = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('/tmp/sm.csv')
df = df.withColumn('verified', F.when(df['ve
浏览 2提问于2016-04-13得票数 23
我目前正在使用Spark 1.6.1，并在Jupyter Ipython笔记本和Java 8中使用它。对于Java，我只需修改我的maven pom以导入Spark 2.0，但我不确定如何在ipython中做同样的事情。我认为我需要安装2.0，但这是可行的，因为我已经安装了Spark 1.6.1？我可以在我的macbook上同时使用两个版本，并从pyspark中选择使用哪个版本吗？多么?
更新:这是我在终端% IPYTHON_OPTS="notebook" pyspark上启动我的Jupyter pyspark笔记本=>的方式。我如何告诉它使用Spark 2.0启动？
浏览 2提问于2016-09-22得票数 0
我的问题是基于的。
是否可能有更详细的注释/解释代码起始行tf = HashingTF().transform( training_raw.map(lambda doc: doc["text"], preservesPartitioning=True))？
我如何打印混淆矩阵？
以下的错误意味着什么？我怎么才能修好它？模型还在建立，我得到了预测
>>> # Train and check ... model = NaiveBayes.train(training) [Stage 2:=============================>
浏览 9提问于2016-04-06得票数 1
1回答
火花流为最后k提供滑动窗口函数，但我想尝试使用切片函数来获得最后k的rdd，在当前时间之前的范围时间内，我想查询rdd。
delta = timedelta(seconds=30)
datates = datamap.slice(datetime.now()-delta,datetime.now())
当执行代码时，我会得到这个错误。
---------------------------------------------------------------------------
Py4JJavaError Traceback (mo
浏览 3提问于2016-01-19得票数 0
我正试着从一份清单中创建一本字典。我有以下清单：
rawPositions
给出
[[1009794, 'LPF6 Comdty', 'BC22', 'Enterprise', 3.0, 3904.125, 390412.5],
[1009794, 'LPF6 Comdty', 'BC22', 'Enterprise', 3.0, 3900.75, 390075.0],
[1009794, 'LPF6 Comdty', 'BC22', 'Enterpris
浏览 3提问于2016-04-22得票数 10
回答已采纳
1回答
我安装了一个带有Hadoop2.6.3和Spark1.6的集群。
最近，我将spark升级到2.0，一切看起来都很好，直到我试着运行一些以前的工作，比如spark 1.6，它与spark 2.0有一些兼容的问题。
我试过的第一件事是：
echo $SPARK_HOME
/usr/local/spark-1.6.1-bin-hadoop2.6
/usr/local/spark-1.6.1-bin-hadoop2.6/bin/spark-submit --master yarn--deploy-mode client /usr/local/spark-1.6.1-bin-hadoop2.6/ex
浏览 3提问于2016-10-21得票数 0
回答已采纳
1回答
我一直在玩spark，但我不能理解如何构建这个执行流。伪代码如下：
from pyspark import SparkConf, SparkContext, SQLContext
sc = SparkContext(conf=conf)
sqlSC = SQLContext(sc)
df1 = getBigDataSetFromDb()
ddf1 = sqlSC.createDataFrame(sc.broadcast(df1))
df2 = getOtherBigDataSetFromDb()
ddf2 = sqlSC.createDataFrame(sc.broadcast(df2))
浏览 1提问于2016-04-19得票数 0
我正在尝试一个简单的网络单词统计程序，使用python中的火花流，代码为
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
sc = SparkContext("local[2]", "NetworkWordCount")
ssc = StreamingContext(sc, 1)
lines = ssc.socketTextStream("localhost", 9999)
words = lines.flatMap(lamb
浏览 2提问于2018-04-10得票数 0
1回答
我正在使用Spark的流API，我只是想更好地了解如何最好地设计代码。
我目前正在使用pyspark.streaming.kafka.createDirectStream的卡夫卡消费者(用火药厂)
根据
火花流提供了一种称为离散流或DStream的高级抽象，它表示连续的数据流。DStreams可以从卡夫卡、弗卢姆和动态等源的输入数据流中创建，也可以通过在其他DStreams上应用高级操作来创建。在内部，DStream被表示为RDDs的序列。
本质上，我希望将一组函数应用于DStream中的每个元素。目前，我正在为pyspark.streaming.DStream使用"map“函
浏览 4提问于2016-02-29得票数 0
我试图向MQTT代理发送一个带有计算结果的DStream，但是foreachRDD一直在崩溃。
我正在运行Spark2.4.3与Bahir的MQTT订阅，从git主编译。到目前为止一切都很顺利。在尝试用MQTT发布我的结果之前，我尝试了saveAsFiles()，这起作用了(但不是我想要的)。
def sendPartition(part):
# code for publishing with MQTT here
return 0
mydstream = MQTTUtils.createStream(ssc, brokerUrl, topic)
mydstream = p
浏览 0提问于2019-07-15得票数 1
回答已采纳
1回答
TL;DR -我在PySpark应用程序中有一个看起来像字符串的DStream。我想将它作为一个DStream[String] 发送到Scala库。但是，Py4j不转换字符串.。
我正在开发一个PySpark应用程序，它使用星火流从Kafka提取数据。我的消息是字符串，我想在Scala代码中调用一个方法，向它传递一个DStream[String]实例。但是，我无法在Scala代码中接收到适当的JVM字符串。在我看来，Python字符串不是转换成Java字符串，而是序列化的。
我的问题是:如何从DStream对象中获取Java？
下面是我想出的最简单的Python代码：
from pyspar
浏览 4提问于2016-09-12得票数 5
回答已采纳
我的错误越来越小。我遵循的是中的线性回归示例。我有spark 1.6.1和python 3.5.1。我应该做哪些更改？
from pyspark.mllib.regression import LabeledPoint, LinearRegressionWithSGD, LinearRegressionModel
# Load and parse the data
def parsePoint(line):
values = [float(x) for x in line.replace(',', ' ').split(' ')]
浏览 0提问于2016-03-18得票数 0
2回答
我试图做一个简单的选择，从别名使用SQLContext.sql在火花1.6。
sqlCtx = SQLContext(sc)
## Import CSV File
header = (sc.textFile("data.csv")
.map(lambda line: [x for x in line.split(",")]))
## Convert RDD to DF, specify column names
headerDF = header.toDF(['header', 'adj', 'des
浏览 8提问于2017-03-07得票数 3
回答已采纳
我刚开始和斯派克一起玩，我已经在挣扎了。我刚刚下载了Spark的spark-1.6.1-bin-hadoop2.4并试图打开PySpark Shell ./bin/pyspark，但不幸的是，我被提示如下：
Error: Could not find or load main class org.apache.spark.launcher.Main
环境：
Ubuntu 14.04.4 64位
Java版本"1.8.0_77“
Python 2.7.6
Spark 1.6.1为Hadoop2.4及更高版本预构建
有什么线索能解决这个问题吗？
，它可以很好地使用S
浏览 1提问于2016-03-31得票数 4
我需要将在线Kafka流媒体期间保存到HDFS的事件注入回DStream PySpark，以进行相同的算法处理。我发现Holden Karau的代码示例“相当于像Kafka那样的可检查点、可重放、可靠的消息队列”。我想知道是否可以在PySpark中实现它：
package com.holdenkarau.spark.testing
import org.apache.spark.streaming._
import org.apache.spark._
import org.apache.spark.rdd.RDD
import org.apache.spark.SparkContext._
浏览 1提问于2018-04-15得票数 0
我有我的spark应用程序，从Kafka读取数据并摄取到Kudu。它已经成功运行了近25个小时，并将数据摄入到了Kudu中。在那之后，我看到从kafka日志中选出了新的kafka分区的领导者。我的应用程序进入完成状态，并出现以下错误
org.apache.spark.SparkException: ArrayBuffer(kafka.common.NotLeaderForPartitionException, org.apache.spark.SparkException: Couldn't find leader offsets for Set([test,0]))
at org.a
浏览 20提问于2018-02-13得票数 0
1回答
我想使用Apache和Spark流来设置一个流应用程序。Kafka运行在一个单独的unix机器版本0.9.0.1上，Sequv1.6.1是hadoop集群的一部分。
我已经启动了动物园管理员和kafka服务器，并希望使用控制台生成器从日志文件中流到消息，并使用直接方法(没有接收者)使用火花流应用程序。我用python编写了代码，并使用以下命令执行：
spark-submit --jars spark-streaming-kafka-assembly_2.10-1.6.1.jar streamingDirectKafka.py
出现以下错误：
/opt/mapr/spark/spark-1.6.
浏览 6提问于2016-08-24得票数 1
回答已采纳
1回答
我在ipython笔记本中使用pyspark并访问netezza数据库。我正尝试在bluemix上做类似的事情。问题是，为了访问netezza，我必须向pyspark启动添加参数。我如何在bluemix上做到这一点？下面是我独立启动pyspark的方法：
$ PYSPARK_DRIVER_PYTHON=ipython PYSPARK_DRIVER_PYTHON_OPTS="notebook" /usr/local/src/spark/spark-1.6.1-bin-hadoop2.6/bin/pyspark --packages com.ibm.SparkTC:spark-ne
浏览 0提问于2016-05-19得票数 0
我使用星火流编写了以下程序
object TrendingHashTags {
def main(args: Array[String]) : Unit = {
val url = getClass.getResource("/twitterapi.properties")
val source = Source.fromURL(url)
val props = new Properties()
props.load(source.bufferedReader())
System.setProperty("twitter4j.
浏览 0提问于2016-05-12得票数 0
回答已采纳
我用的是Windows8.1个人电脑。Spark和python版本信息如下
有没有办法将Spark设置为使用python2而不是3.5.1？
c:\spark-1.6.1-bin-hadoop2.6\spark-1.6.1-bin-hadoop2.6>bin\pyspark
Picked up _JAVA_OPTIONS: -Djava.net.preferIPv4Stack=true
Python 3.5.1 |Anaconda 2.5.0 (64-bit)| (default, Jan 29 2016, 15:01:46) [MSC v.1900 64 bit (AMD64)] on
浏览 0提问于2016-03-21得票数 0
1回答
我是spark流媒体的新手。
当我像普通scala应用程序一样运行spark streaming功能时，它可以正常工作。
我能够捕获我的kafka事件，并能够存储在hdfs本身中。
当我尝试使用spark-submit命令作为我的流jar运行时，我得到了下面的错误。
spark-submit --class Test --master yarn --executor-memory 20G --num-executors 50 spark-kafka-streaming-0.0.1-SNAPSHOT-jar-with-dependencies.jar
16/02/16 08:39:23
浏览 1提问于2016-02-17得票数 1
1回答
我正在尝试连接spark streaming应用程序中的DB2数据库和数据库查询执行语句，这会导致"org.apache.spark.SparkException:任务不可序列化“问题。请给我建议。下面是我的示例代码以供参考。
dataLines.foreachRDD{rdd=>
val spark = SparkSessionSingleton.getInstance(rdd.sparkContext.getConf)
val dataRows=rdd.map(rs => rs.value).map(row =&g
浏览 1提问于2018-03-29得票数 0
我正在尝试从S3中将大约一百万个压缩文本文件读入spark。每个文件的压缩大小在50 MB到80 MB之间。总共大约有6.5万亿字节的数据。
不幸的是，我遇到了一个内存不足的异常，我不知道如何解决。像这样简单的东西：
raw_file_list = subprocess.Popen("aws s3 ls --recursive s3://my-bucket/export/", shell=True, stdout=subprocess.PIPE).stdout.read().strip().split('\n')
cleaned_names = ["s3
浏览 20提问于2016-07-26得票数 4
回答已采纳
2回答
我是Spark Streaming的新手，我正在尝试使用pyspark读取和解析来自Kafka的JSON流。读取流是可以的，而且我也可以pprint() RDDs。
{"Address":"22.79.52.79","AlarmProfile":"-1","Amps":"11.98","AmpsLimit":"90","AssetTag":"-1","AssetTag_1":"-1","Blan
浏览 3提问于2016-03-04得票数 1
2回答
试图将星星之火数据存储为paraquet file.But无法实现，由于异常below.Kindly指导我，如果我缺少something.The数据save已经从kafka流rdds构建。
dataframe.write.paraquet("/user/space")
异常堆栈：
Exception in thread "streaming-job-executor-0" java.lang.NoClassDefFoundError: org/apache/spark/sql/execution/datasources/FileFormat
at java.la
浏览 5提问于2016-10-31得票数 2
回答已采纳
2回答
我在使用PySpark的spark上有一个不平衡的数据帧。我想对其进行重新采样，使其达到平衡。我只在PySpark中找到了示例函数
sample(withReplacement, fraction, seed=None)
但是我想在Python中用单位体积的权重来采样数据帧，我可以这样做
df.sample(n,Flase,weights=log(unitvolume))
有没有什么方法可以用PySpark做同样的事情呢？
浏览 2提问于2018-02-01得票数 5
下面是一个长期运行的星火流应用程序的例外情况。异常可能在几分钟后发生，但也可能在几天内不会发生。这是与相当一致的输入数据。
我见过，但我不认为这是同一个问题。那是java.lang.IllegalArgumentException，这是java.io.IOException: Class not found。
我的应用程序是流数据和使用Spark写入Parquet。
我用的是星火1.5.2。有什么想法吗？
28-01-2016 09:36:00 ERROR JobScheduler:96 - Error generating jobs for time 1453973760000 ms
jav
浏览 2提问于2016-01-28得票数 6
回答已采纳
因为我接收到来自服务器不同的StreamingContext.union topics.So的消息，所以我需要使用streams.But方法来合并streams.But，我在尝试将streams.But偏移更新到动物园管理员时遇到了一些问题。
错误如下：
java.lang.ClassCastException: org.apache.spark.rdd.UnionRDD cannot be cast to org.apache.spark.streaming.kafka.HasOffsetRanges
at com.qingqing.spark.util.KafkaManager.updat
浏览 3提问于2017-06-24得票数 0
我在Python中使用Spark3.0.0。我有一个卡夫卡的test_topic，正在生产从一个csv。
下面的代码是从这个主题消耗到Spark中的，但是我在某个地方读到它需要在DStream中才能对它执行任何ML。
import json
from json import loads
from kafka import KafkaConsumer
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
sc = SparkContext("local[2]", "
浏览 5提问于2020-08-06得票数 1
回答已采纳
1回答
我只是试着运行statefu的示例代码，但是它失败了。无法了解为什么会发生这种情况。
在cloudera vm 5.13.3上使用3.6 python火花2.3
运行选项：
--master local[*] --queue PyCharmSpark pyspark-shell
我的代码是：
from pyspark import SparkConf, SQLContext
from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext
from pyspark.sql.functions
浏览 2提问于2018-08-27得票数 0
我刚开始在Python中使用Spark，并且一直无法解决这个问题:在pyspark.sql.dataframe.DataFrame上运行groupBy之后
df = sqlsc.read.json("data.json")
df.groupBy('teamId')
如何从每个结果组(按teamId分组)中选择N随机样本而不进行替换？
我基本上是尝试从每个团队中随机选择N用户，也许一开始使用groupBy是错误的？
浏览 0提问于2015-11-17得票数 21
回答已采纳
显而易见的是，我最近在UBUNTU (VMWARE工作站)上安装了SPARK。下面是我的电脑规格。
Windows Dell笔记本电脑(运行windows 10)。
已安装VMWARE PRO 12并在其上加载了Ubuntu 15。使用标准脚本安装SPARK 1.6.1 / JAVA 1.7 / Python 2.7和SCALA 2.11.8。
我使用spark submit命令运行了一个示例程序，它完成得很好。但是当我尝试登录pyspark shell时，我得到了错误信息"pyspark: command not found“
似乎是什么问题。我可以看到spark的bin目录中的所有
浏览 0提问于2016-05-07得票数 0
#Load the CSV file into a RDD
irisData = sc.textFile("/home/infademo/surya/iris.csv")
irisData.cache()
irisData.count()
#Remove the first line (contains headers)
dataLines = irisData.filter(lambda x: "Sepal" not in x)
dataLines.count()
from pyspark.s
浏览 2提问于2017-04-21得票数 0
val file = File.createTempFile("temp", ".avro")
val schema = new Schema.Parser().parse(st)
val datumWriter = new GenericDatumWriter[GenericData.Record](schema)
val dataFileWriter = new DataFileWriter[GenericData.Record](datumWriter)
dataFileWriter.create(schema , file)
rdd.foreach(r
浏览 0提问于2017-06-13得票数 1
1回答
我正在尝试使用eclipse IDE中的java代码通过kafka producer生成一些随机数据。我在kafka consumer中收到了相同的数据，它也是在同一个IDE中使用java代码创建的。我的工作依赖于流数据。所以，我需要spark streaming来接收kafka生成的随机数据。对于spark streaming，我在jupyter-notebook中使用python代码。要集成kafka和spark，必须在spark jar中添加“spark-streaming kafka-0-10_2.12-3.0.0.jar”文件。我还尝试在pyspark中添加jar文件。这是我的spa
浏览 0提问于2020-11-09得票数 1
到目前为止，Spark还没有为流数据创建DataFrame，但是当我进行异常检测时，使用DataFrame进行数据分析更方便、更快。我已经做了这一部分，但是当我尝试用流式数据进行实时异常检测时，问题就出现了。我尝试了几种方法，但仍然无法将DStream转换为DataFrame，也无法将DStream中的RDD转换为DataFrame。
下面是我最新版本的代码的一部分：
import sys
import re
from pyspark import SparkContext
from pyspark.sql.context import SQLContext
from pyspark.sql
浏览 2提问于2016-02-06得票数 10
回答已采纳
我用pyspark写了一个Spark Streaming程序。
它通过socketTextStream接收实时输入的文本流，并进行相应的转换，然后通过saveAsTextFile将其保存为csv文件。不使用Spark streaming窗口操作，并且不需要以前的数据来创建输出数据。
但是似乎在前一个RDD完成之前，Spark不会开始处理DStream中的RDD，即使前一个RDD只使用了几个分区和CPU/内存。
这是Spark的默认行为吗?有什么方法可以改变这种行为吗？
浏览 1提问于2015-04-29得票数 0
我有一个由list对象组成的RDD对象(示例：[1,2,3,4,5,6] )。我需要对其进行采样，并为此使用sample() pyspark函数，但它看起来检索的是第一个元素，而不是随机采样。 我的想法是将元素打乱，然后使用sample()函数，但我无法进行随机化。
浏览 120提问于2021-04-15得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
