在Spark structured Streaming中，我想从STRING创建一个StructType。在下面的示例中，spark read方法只接受schema的"Struct Type“，我如何从字符串创建StructType。我想要将employeeSchema字符串转换为StructType。EmployeeSchemaLoa
浏览 15提问于2017-08-31得票数 1
我创建了一个带有JSON格式记录的Kafka主题。.from-beginning \ --consumer.config /root/client.properties
我如何使用Python中的Spark流来实现这一点？
浏览 5提问于2017-05-24得票数 1
回答已采纳
我有一个关于Kafka流上的Spark结构化流媒体的问题。我有一个模式，类型是： .add("field1", StringType)这是我的导入：import org.apache.spark.sql.Row;
import org.apache.<e
浏览 10提问于2018-09-24得票数 0
我使用的是spark sturctured (2.3)和kafka 2.4版本。import org.apache.spark.sql.catalyst.encoders.RowEncoder;
import org.ap
浏览 0提问于2020-03-20得票数 0
我正在浏览下面的博客中的spark structured。
.add("Records", ArrayType(new StructType()
.add("additionalEventData例如，在spark batch streaming中，我们在下面的代码行中<
浏览 3提问于2018-03-23得票数 1
我从Kafka那里得到了JSON格式的数据，并在PySpark中以DataFrame的形式读取了这些数据。在我从Kafka获得数据后，它显示为DataFrame格式： DataFrame[value: string] 但是，该值包含JSON / DICT格式。
浏览 24提问于2020-01-13得票数 0
2回答
我需要一个应用程序，使工人之间的通信。假设worker 1正在处理job 1，它将生成其他works所依赖的数据输出。此外，这个过程应该重复多次，这意味着每当worker 1生成新的数据集时，其他worker都应该开始输入该数据集并完成自己的工作。spark能做到吗？到目前为止，我已经看到了spark流实时处理，但流通信似乎没有发生在工人之间？任何方向或建议都将不胜感激。
浏览 1提问于2015-07-06得票数 0
我正在尝试使用pyspark插入kafka流中的数据 cols = ['id','name'] # spark contextpara_seconds) brokers = 'localhost:9092' # get <
浏览 1提问于2021-06-30得票数 0
我正在写一个自定义的Spark流媒体源码。我想支持列修剪。{
schema = requiredSchema我正在使用模式创建批处理行:我已经检查过，在我返回的<
浏览 2提问于2018-06-29得票数 2
我是星火流的新手。我试图实现的是从kafka读取json字符串数据，将其存储在DStream中，并将其转换为Dataset，以便将其加载到Elasticsearch中。我正在使用这个的部分代码。; import org.apache.spark.sql.types.StructType<
浏览 1提问于2017-05-05得票数 0
回答已采纳
我使用的是apsche spark streaming 2.3.1，其中我接收到一个流，其中包含"HH:mm:ss.xxxxxxxxx“格式的时间戳值(13:09:05.761237147)作为字符串。我需要将此字符串转换为timestamp数据类型。spark = SparkSession \ .appName("abc") \
schema = Str
浏览 0提问于2018-09-04得票数 2
3回答
我使用Spark从文件夹中读取文本文件，并将它们加载到hive中。我如何控制火花流，以限制程序读取的文件数量？目前我的程序正在读取过去1分钟内生成的所有文件。但我想控制它正在读取的文件数量。
我使用的是textFileStream接口。
浏览 4提问于2017-02-28得票数 0
1回答
val df = spark.read.avro(file)尝试手动创建架构，但现在遇到以下问题：
val df = spark.read$I
浏览 0提问于2018-06-15得票数 1
我试图从一个特定的文件夹中读取CSV文件，并将相同的内容写入到本地pc上不同位置的其他CSV文件中，以供学习。我可以读取文件并在控制台上显示内容。我在这里一步一步地粘贴整个代码：spark = SparkSession \.appName('csv01') \
spark.conf.set(&qu
浏览 3提问于2018-06-09得票数 0
回答已采纳
我正在尝试从Amazon SQS队列中读取消息。权限正在工作，我可以看到记录计数-但所有记录都为空。我不知道为什么我会得到空值。我可以看到SQS队列中的消息，也可以从本地Python实例中获取它们，它们实际上是反映模式的JSON记录(尽管不能百分之百确定我是否正确地实现了模式)。import org.
浏览 3提问于2018-05-03得票数 0
我遵循了的通用Spark测试问题，我最接近的尝试是[]，看起来像这样：import org.scalatest.FunSpec{StringType, IntegerType, DoubleType, StructType, DateType}
it("Rea
浏览 1提问于2018-03-28得票数 6
1回答
我使用的是Spark 1.6和Kafka 0.8.2.1。
我正在尝试使用Spark Streaming从Kafka获取一些数据，并对这些数据进行一些操作。为此，我应该知道获取的数据的模式，有什么方法可以做到这一点，或者我们可以通过使用字段名从流中获取值吗？
浏览 0提问于2017-05-29得票数 1
2回答
当我试图将星星之火应用程序分解成类并使用Try时，我遇到了一个无法串行化的任务。我试过用这门课而不是用这门课..。但在这两种情况下，我得到了一个与闭包相关的serz错误。我猜是有东西被牵扯进来了。这个错误一直困扰着我..。四处走动真是太痛苦了。如果有人能对我如何避免这个问题有所了解，那就太棒了。这些Java类似乎存在比有时更多的问题。{col, from_json}
import org.apache.spark</
浏览 1提问于2018-07-30得票数 1
回答已采纳
我有一个错误：
org.apache.spark.api.java.function.VoidFunction2[org.apache.spark.sql.Datasetorg.apache.spark.sql.Row，java.lang.Long])org.apache.spark.sql.streaming.DataStreamWriterorg.apache.spark.sql.streaming.DataStreamWriterorg.apache.sp
浏览 5提问于2020-07-28得票数 6
2回答
火花流写问题
、、
我试图从文本文件中的JSON记录中创建一个动态模式，因为每个记录都有不同的模式。下面是我的代码。import org.apache.spark.sql.SparkSessionimport org.apache.spark.sql
浏览 5提问于2020-03-10得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
