我正在使用pyspark读取Kafka的流数据，然后我想将这些数据汇到mongodb。我已经包含了所有必需的包，但是它会引发以下错误
UnsupportedOperationException:数据源com.mongodb.spark.sql.DefaultSource不支持流写入。
以下链接与我的问题无关
下面是完整的错误堆栈跟踪
回溯(最近一次调用)：.option("com.mongodb.spark.sql.DefaultSource"，“mongodb://localhost:27017/twitter.test”中的文件“/home/b3ds/kafka-sc
浏览 4提问于2018-06-04得票数 2
1回答
如何计算或管理火花放电中的流数据
、、、、
我想从流数据中计算数据，然后发送到网页。例如，：我将在流数据中计算TotalSales列的和。，但它在summary = dataStream.select('TotalSales').groupby().sum().toPandas()上出错，这是我的代码。
import os
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType
from pyspark.sql.functions import *
spark = SparkSession.builder.appName
浏览 0提问于2020-06-02得票数 0
回答已采纳
1回答
我有来自卡夫卡主题的数据。在我的dataframe中，我有“平均”列，我希望通过对“平均”列执行一些计算来创建新列。
目前我写的是-
rdd_get_chills = df_avg_tmp.rdd.map(lambda line:get_wind_chills(line))
但这是个错误-
pyspark.sql.utils.AnalysisException: Queries with streaming sources must be executed with writeStream.start();
浏览 3提问于2022-09-20得票数 0
回答已采纳
1回答
我想对我从一个卡夫卡集群中获得的消息流执行tweet情绪分析，该集群反过来从Twitter v2中获取这些消息。
当我尝试应用预先训练过的情感分析管道时，我会收到一条错误消息，上面写着：Exception: target must be either a spark DataFrame, a list of strings or a string，我想知道是否有办法解决这个问题。
我已经检查了文档，在流数据上找不到任何东西。
这是我使用的代码：
import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.functi
浏览 7提问于2022-03-23得票数 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流从kafka获取数据，结果是一个dataframe，当我将dataframe转换为rdd时，它出错了： Traceback (most recent call last):
File "/home/docs/dp_model/dp_algo_platform/dp_algo_core/test/test.py", line 36, in <module>
df = df.rdd.map(lambda x: x.value.split(" ")).toDF()
File "/home/softs/spark-2.4.3-
浏览 56提问于2020-01-06得票数 3
我有一个Kafka2.1消息代理，希望对Spark2.4中的消息数据进行一些处理。我想用齐柏林飞艇0.8.1笔记本进行快速原型。
我下载了结构化流()所必需的星火流-kafka-0-10_2.11.jar，并将其作为“依赖-工件”添加到齐柏林飞艇( Zeppelin )的“streaming”-interpreter(也处理%pyspark段落)中。我重新启动了这个解释器(还有齐柏林飞艇)。
我还把罐子装进了笔记本的第一段(我首先认为这不应该是必要的.)：
%dep z.load("/usr/local/analyse/jar/spark-streaming-kafka-0-10_2.
浏览 1提问于2019-07-19得票数 5
回答已采纳
我正在尝试测试一个实用函数，它接受一个spark DataFrame，并在经过一些转换后输出一个新的spark DataFrame。在尝试模拟测试数据时，我需要为输入构造一个pyspark dataframe。我看到的大多数使用这种用法的例子 spark.createDataFrame(data, columns) 我对文档不太熟悉，找不到"spark“。您如何使用from pyspark* import spark
浏览 8提问于2020-08-12得票数 0
回答已采纳
我已经设置了一个Kafka代理，并且我设法用pyspark读取记录。
import os
from pyspark.sql import SparkSession
import pyspark
import sys
from pyspark import SparkConf, SparkContext, SQLContext, Row
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
conf = SparkConf().setMaster("my
浏览 12提问于2020-05-20得票数 0
我试图使用保存的Mllib模型来预测实时流数据的情绪。
我已经试过了我找到的所有建议，但我还是会犯错误。当前错误:字段“功能”不存在。
经过训练的数据模式是
root
|-- label: double (nullable = true)
|-- words: array (nullable = true)
| |-- element: string (containsNull = true)
|-- features: vector (nullable = true)
lines = spark\
.readStream\
.format(
浏览 0提问于2019-11-07得票数 1
每当我试图转换它时，这就是我得到的例外。
val df_col = df.select("ts.user.friends_count").collect.map(_.toSeq)
org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;
我所要做的就是在结构化流中复制以下sql.dataframe操作。
df.collect().foreach(row => droolsCaseClass(row.ge
浏览 0提问于2018-01-11得票数 3
回答已采纳
我对PySpark结构化流有以下问题。
流数据中的每一行都有一个用户ID和一个时间戳。现在，对于每一行和每个用户，我想添加一个带有时间戳差异的列。
例如，假设我收到的第一行是："User A，08:00:00“。如果第二行显示"User A，08:00:10“，那么我想在第二行中添加一个名为"Interval”的列，表示"10秒“。
有没有人知道如何做到这一点？我尝试使用结构化流文档的窗口函数示例，但是没有用。
非常感谢
浏览 0提问于2019-11-14得票数 1
我试着用spark连接到kafka的话题。它不会读取数据流中的任何数据，也不会产生任何错误。下面是我的jupyter代码：
import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2 pyspark-shell'
from pretty import pprint
from pyspark import SparkContext
from pyspark.streaming import Stream
浏览 0提问于2019-12-19得票数 1
我想使用星火流，并连接它与卡夫卡。然而，我仍然得到了NoSuchMethodError: NoSuchMethodError错误，现在我只是不知道下一步该做什么。
我的设置：
Ubuntu 16.04
Scala 2.11
Kafka 2.11-1.0.0 (我也尝试使用2.11-0.10.0.0)
火花2.2.1
Hadoop 2.9.0
我甚至不能运行示例脚本：
from pyspark.sql import SparkSession, Row, SQLContext
from pyspark import SparkContext, SparkConf
from pyspark.stre
浏览 1提问于2018-02-09得票数 3
3回答
当我从Kafka主题创建一个流并打印它的内容时
import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2 pyspark-shell'
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka impo
浏览 2提问于2019-12-08得票数 1
回答已采纳
我试图每5分钟读取一次csv文件，并输出不到5分钟前添加到另一个csv文件中的任何新行。我认为一个好的垫脚石是readStream一个csv文件，writeStream用于控制台，每当我手动将一行数据添加到csv文件中时，就打印出文件内容。问题是，只有当我在目录中插入一个全新的csv文件时，控制台才会打印。
当我的新行被添加到csv文件时，我如何打印？
这是我的目录布局：
stream_script2.py (代码below)csv_files unit_testing.csv )
from pyspark.sql import SparkSession
from pyspark.sql.ty
浏览 3提问于2020-09-29得票数 0
目前，我正在使用火花结构化流创建随机数据的数据帧(id，timestamp_value，device_id，temperature_value，注释)。
每批火花数据采集：
基于上面数据框架的屏幕截图，我想为列"temperature_value“提供一些描述性的统计数据。例如，最小、最大、平均、计数、方差。
我在python中实现这一目标的方法如下：
import sys
import json
import psycopg2
from pyspark import SparkContext
from pyspark.streaming import StreamingCont
浏览 4提问于2019-12-17得票数 2
回答已采纳
我正在使用pandasUDF将标准的ML python库应用于pyspark DataFrame。在定义了模式并进行了预测之后，我得到了pyspark DF作为输出。现在，我想用这个预测数据帧做一些事情，例如，我尝试对列"weekly_forecast_1“中的所有值进行求和。当我应用.collect()或.toPandas()方法时，在.fit()中得到以下错误 IndexError: too many indices for array:array is 0-dimensional, but 1 were indexed 每当我尝试将.collect()或.toPandas()方
浏览 33提问于2020-11-24得票数 0
1回答
我使用Spark2.2.0。
如何使用pyspark向Amazon流提供火花结构化流？
问题试图通过创建自定义接收器来回答非结构化流和scala的问题。
火花放电也有类似的可能吗？
spark.readStream \
.format("s3-sqs") \
.option("fileFormat", "json") \
.option("queueUrl", ...) \
.schema(...) \
.load()
根据的要求，接收机可以作为S3-SQS的文件源.不过，只有SQS才能采用一种方法。
浏览 3提问于2017-12-28得票数 15
回答已采纳
我正在尝试应用一个用Python语言编写的函数，以便在PySpark上可读。 open()函数在PySpark中不可读。在这种情况下如何更换？ 我有一个需要排序的非结构化列表： import csv
from operator import itemgetter
my_list = [
[3, 'ab','ac','ad'],
[4, 'ae','af','at','aj','aa'],
[1, 'ar'
浏览 29提问于2021-10-15得票数 0
1回答
导入KafkaUtils时，我遇到了问题:没有名为'pyspark.streaming.kafka‘的模块，但我不知道如何安装kafka模块。
我使用python 3.6.8，spark 2.2.0和kafka_2.12-2.5.0
浏览 0提问于2020-07-06得票数 1
我正在尝试将pandas udf应用于pyspark结构化流的窗口。问题是，一旦流赶上了当前状态，所有新窗口都只包含一个值。
如截图所示，2019-10-22T15:34:08.730+0000之后的所有窗口都只包含一个值。用于生成此代码的代码如下：
@pandas_udf("Count long, Resampled long, Start timestamp, End timestamp", PandasUDFType.GROUPED_MAP)
def myudf(df):
df = df.dropna()
df = df.set_index("T
浏览 3提问于2019-10-23得票数 1
1回答
我在databricks notebook (python)中有以下流代码。 from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
from pyspark.sql.functions import split
spark = SparkSession \
.builder \
.appName("MyTest") \
.getOrCreate()
# Create a streaming DataFrame
lines = spark.read
浏览 21提问于2021-04-23得票数 1
回答已采纳
我有一个要求，把从火花放电脚本创建的日志推到kafka。我正在做POC，所以在windows机器上使用Kafka二进制文件。我的版本是- kafka - 2.4.0，火花- 3.0和python-3.8.1。我用的是吡喃编辑器。
import sys
import logging
from datetime import datetime
try:
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka im
浏览 3提问于2020-02-12得票数 3
回答已采纳
1回答
我有大约500-700 csv文件的日志加载每周。文件名是每周随机生成的。这些文件存储在google云存储上，pyspark在google dataproc上运行，有几个工作节点。
我已经创建了一个pyspark文件，它基本上只是迭代文件并按原样加载到bigquery中。我以为我会在bigquery表上看到增量加载，但似乎pyspark会将整个gz输入文件合并到dataframe中，然后立即处理它。所以我在大约90分钟后得到了结果。
我如何编写pyspark代码，以便每次读取文件时，都会将其写入bigquery。总体时间可能不会有太大变化，但至少我如何才能看到部分写入bigquery的数据？
浏览 22提问于2021-10-10得票数 1
1回答
带有火花流的Kafka抛出了一个错误：
from pyspark.streaming.kafka import KafkaUtils ImportError: No module named kafka
我已经建立了一个卡夫卡经纪人和一个工作火花环境与一个主人和一个工人。
import os
os.environ['PYSPARK_PYTHON'] = '/usr/bin/python2.7'
import findspark
findspark.init('/usr/spark/spark-3.0.0-preview2-bin-hadoop2.7&#
浏览 2提问于2020-05-19得票数 4
回答已采纳
在python或R中，可以使用索引对DataFrame进行切片。
例如，在熊猫中：
df.iloc[5:10,:]
在pyspark中有没有类似的方式来根据行的位置对数据进行切片？
浏览 3提问于2018-10-13得票数 4
回答已采纳
有人能建议我如何在pyspark dataframe中添加多个空列吗？目前我正在做这样的事情，但它不起作用：
def add_columns(dataframe, column_list):
for col in column_list:
self = dataframe.withColumn(str(col), lit(None).cast(StringType())))
return dataframe
在应用add_columns函数后的输出模式中，我在0x7f41189d7e10处获得了新的列作为生成器对象genexpr : string (nullab
浏览 1提问于2018-09-14得票数 0
在中，它说：
Kafka流应用程序只能与此配置值指定的单个Kafka集群通信。未来版本的Kafka流将支持连接到不同的Kafka集群来读取输入流和写入输出流。
这是否意味着我的整个应用程序只能连接到一个卡夫卡集群，或者每个实例只能连接到一个集群？
我是否可以创建多个KafkaStreams实例，它们具有连接到不同集群的不同属性？
浏览 10提问于2017-08-23得票数 13
1回答
我对kafka非常陌生，我试图将数据写到一个主题中，并从同一个主题中读取数据(我们现在作为一个源团队来摄取数据)。因此，我们在对Kafk主题进行写操作，并从相同的主题中使用)。我在星火壳上编写了下面的代码，以便将数据写入一个Kafka主题。
pyspark --packages io.delta:delta-core_2.11:0.6.1,org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.0,io.strimzi:kafka-oauth-client:0.5.0
from pyspark.sql.functions import col
from
浏览 3提问于2021-09-08得票数 1
回答已采纳
对于ApacheSparkVersion2.1，我想使用Kafka (0.10.0.2.5)作为结构化流的源：
kafka_app.py：
from pyspark.sql import SparkSession
spark=SparkSession.builder.appName("TestKakfa").getOrCreate()
kafka=spark.readStream.format("kafka") \
.option("kafka.bootstrap.servers","localhost:6667") \
.o
浏览 4提问于2017-03-20得票数 2
回答已采纳
我想使用PySpark结构流应用编程接口将结构流数据写入Cassandra。
我的数据流如下：
REST API -> Kafka ->火花结构流(PySpark) -> Cassandra
源代码和版本如下: Spark版本: 2.4.3 DataStax DSE: 6.7.6-1
初始化spark：
spark = SparkSession.builder\
.master("local[*]")\
.appName("Analytics")\
.config("kafka.bootstrap.servers", "
浏览 23提问于2019-11-01得票数 2
1回答
我已经为pyspark dataframe中的两个特性应用了groupby和计算标准差。 from pyspark.sql import functions as f
val1 = [('a',20,100),('a',100,100),('a',50,100),('b',0,100),('b',0,100),('c',0,0),('c',0,50),('c',0,100),('c',0,20)]
cols = ['group',
浏览 18提问于2019-04-15得票数 0
我想使用结构化流和PySpark一起运行Spark应用程序。
我使用星火2.2和卡夫卡0.10版本。
如果出现以下错误，我将失败：
java.lang.IncompatibleClassChangeError:实现类
spark-submit命令，如下所示：
/bin/spark-submit \
--packages org.apache.spark:spark-streaming-kafka-0-10_2.11:2.2.0 \
--master local[*] \
/home/umar/structured_streaming.py localhost:2181 fo
浏览 5提问于2017-08-22得票数 3
假设您使用一个精确的模式创建了一个火花DataFrame：
import pyspark.sql.functions as sf
from pyspark.sql.types import *
dfschema = StructType([
StructField("_1", ArrayType(IntegerType())),
StructField("_2", ArrayType(IntegerType())),
])
df = spark.createDataFrame([[[1, 2, 5], [13, 74, 1]],
浏览 11提问于2022-11-29得票数 0
回答已采纳
1回答
我正在尝试将矩阵的平面文本文件(以制表符分隔)表示到dask数组中，使用distributed将数组的块分布到集群中。
(旁白:这与PySpark的方法没有什么不同)
但是，我不清楚如何使用可用的工具将文本文件解析成矩阵结构。在PySpark中，我可以使用一个简单的map来完成这个任务，该map可以将空格上的每一行拆分，并将令牌转换为浮动列表，最后将列表转换为索引的numpy数组。
我从dask.bag.read_text方法开始，它有一个参数collection，可以设置为False，因此它给了我一堆Delayed的未来。如何将这些解析成最终矩阵的行？
from dask.bag impor
浏览 1提问于2016-06-03得票数 1
回答已采纳
我正在使用twitter流函数，它提供了一个流。我需要使用星火writeStream函数，如：
// Write key-value data from a DataFrame to a specific Kafka topic specified in an option
val ds = df
.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.writeStream
.format("kafka")
.option("kafka.bootstrap
浏览 2提问于2019-11-13得票数 1
回答已采纳
我正在寻找一个包，或者使用redshift作为结构化数据流的源的以前的实现。
spark.readStream \
.format("io.github.spark_redshift_community.spark.redshift") \
.option('url', redshift_url) \
.option('forward_spark_s3_credentials', 'true') \
.load()
使用下面的格式，您会在读取时出错。例如：
Data source io.github
浏览 2提问于2020-11-18得票数 0
1回答
我对pySpark非常陌生，但我对火花数据采集(例如，由RDD创建)和一只在星火中的熊猫之间的区别感到困惑。
那是同一个物体吗？从类型上看，它们似乎是不同的类。
核心的区别是什么，如果有的话？(我知道，与熊猫在星星之火中工作，你可以在分布式的Dataframe上使用Pandas的几乎相同的语法，但我想知道是否只有这一个才是不同的)
谢谢
浏览 2提问于2022-02-04得票数 6
1回答
使用SSL连接到Kafka集群的Python (pyspark)的正确语法是什么？
我有一个包含SSL的本地kafka集群(用于测试目的)。我能够将Scala与本地集群连接起来，但还不能连接PySpark。
到目前为止，我有：
kafkaBrokers = "localhost:9093"
schemaRegistryUrl = "https://registry:8081/"
inputTopic = "test.spark"
df = spark.readStream.format("kafka").option("k
浏览 0提问于2018-04-19得票数 0
2回答
我在DataFrame中有3列：- time:TimeStamp，col1:Double，col2:Double我想执行以下操作：
dataFrame.withWatermark("time", "10 seconds")
.groupBy(window(col("time"),"10 seconds","1 second"))
.agg(mean("col1") with window of 10 seconds,max("col") with
浏览 0提问于2017-08-30得票数 1
您好，我正在尝试使用pyspark和kafka创建一个neo4j接收器，但由于某些原因，此接收器在neo4j中创建了重复项，我不确定为什么会发生这种情况。我希望只得到一个节点，但看起来它正在创建4个节点。如果有人有想法，请让我知道。
Kafka生产者代码：
from kafka import KafkaProducer
import json
producer = KafkaProducer(bootstrap_servers='10.0.0.38:9092')
message = {
'test_1': 'test_1',
&
浏览 0提问于2020-03-20得票数 1
在中，我尝试在笔记本中创建一个kafka流，并使用它来创建一个火花作业。Databricks在KafkaUtils.createDirectStream()行抛出错误。在下面附加了相应的代码。
from kazoo.client import KazooClient
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils, TopicAndPartition
sc = spark.sparkContext
ssc = StreamingContext(sc, 3
浏览 8提问于2019-11-21得票数 2
回答已采纳
2回答
我是作为JSON从kafka_2.11-0.10.0.1到Spark2.1的流式计量器读取记录。我转向了结构化流媒体；尽管卡夫卡用户确认了输入的数据，但我的控制台和writeStream没有移动。我正在测试使用
pyspark --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.1.0
我的代码：
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
spark = SparkSe
浏览 0提问于2017-12-15得票数 3
回答已采纳
我在PySpark中这样做过：
使用DataFrame语句创建一个SELECT，以获取按资产序列号排序的资产数据，然后按时间排序。
使用DataFrame.map()将DataFrame转换为RDD。
使用整理每个资产的所有数据，使用资产的序列号作为密钥。
问题：，我能确定每个资产的数据仍然会在最后一步的RDD中按时间顺序排序吗？
时间顺序对我来说是至关重要的(我需要在每个资产的移动时间窗口上计算统计数据)。当RDD.combineByKey()为给定的密钥组合来自火花集群中不同节点的数据时，该键的数据中是否保留了任何顺序？或者，对于给定的密钥，来自不同节点的数据是否按特定
浏览 1提问于2017-04-26得票数 0
回答已采纳
1回答
我目前正在学习使用RabbitMQ。我正在尝试使用pyspark从向RabbitMQ发布一条消息。你知道怎样才能做到吗？
浏览 0提问于2019-12-05得票数 0
回答已采纳
我有一个PySpark DataFrame -
valuesCol = [('Sweden',31),('Norway',62),('Iceland',13),('Finland',24),('Denmark',52)]
df = sqlContext.createDataFrame(valuesCol,['name','id'])
+-------+---+
| name| id|
+-------+---+
| Sweden| 31|
| Norway| 62|
|Icela
浏览 1提问于2018-10-29得票数 17
回答已采纳
3回答
寻找一些信息使用自定义分区在火花公子。我有一个数据存储不同国家的国家数据。因此，如果我对country列进行重新分区，它将将我的数据分发到n个分区中，并将类似的国家数据保存到特定的分区中。当我看到使用glom()方法时，这将创建一个倾斜的分区数据。
一些国家，如美国和中国，有大量的数据，特别是数据。我想重新划分我的数据，如果国家是美国和CHN，那么它将进一步分裂为大约10个分区，其他保持相同的其他国家，如印度，THA，AUS等。
我在下面的链接中看到了这样的内容:我们可以在Spark应用程序中扩展scala分区器类，并可以修改partitioner类以使用自定义逻辑根据需求重新划分我们的数据。
浏览 3提问于2018-10-13得票数 6
回答已采纳
2回答
火花结构化流处理每一行
、、、
我使用结构化流与Spark2.1.1。我需要将一些业务逻辑应用于传入消息(来自Kafka源)。
本质上，我需要获取消息，获取一些键值，在HBase中查找它们，并在dataset上执行更多的biz逻辑。最终结果是需要写入另一个Kafka队列的字符串消息。
但是，由于传入消息的抽象是dataframe (无限制的表结构流)，我必须通过mapPartitions (由于HBase客户端不可串行化而导致的分区)在触发器期间迭代所接收的数据集。
在我的过程中，我需要迭代每一行来执行相同的业务流程。
是否有更好的方法可以帮助我避免dataFrame.mapPartitions调用？我感觉到它的顺序和迭
浏览 9提问于2017-05-25得票数 3
3回答
我有一个结构化的流查询，它深入到Kafka。此查询具有复杂的聚合逻辑。
我想将这个查询的输出DF汇到多个Kafka主题，每个主题在不同的‘key’列上分区。我不希望每个不同的Kafka主题都有多个Kafka接收器，因为这意味着要运行多个流查询--每个Kafka主题一个，特别是因为我的聚合逻辑很复杂。
问题：
是否有一种方法可以将结构化流查询的结果输出到多个Kafka主题，每个主题都有不同的键列，但不必执行多个流查询？
如果没有，那么将多个查询级联起来，使第一个查询进行复杂聚合并将输出写入Kafka，然后其他查询只读取第一个查询的输出并将其主题写入Kafka，从而避免再次进行复杂的聚合，
浏览 1提问于2018-02-13得票数 5
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
