我在kafka流中有两个主题，我想在一些RDD集上执行一个转换，在其余的RDD上执行另一个转换。下面是在spark流中获取kafka数据的代码 KafkaUtils.createDirectStreamkafka.serializer.StringDecoder.class, kaf
浏览 0提问于2017-11-28得票数 0
回答已采纳
我目前正在自学Spark programming，并试图用PySpark重新编写一个现有的Python应用程序。然而，我仍然对如何在PySpark中使用常规Python对象感到困惑。我了解Spark中的分布式数据结构，如RDD、DataFrame、Datasets、vector等。Spark有自己的转换操作和动作操作，如.map()、.reduceByKey()来操作这些对象。但是
浏览 34提问于2017-03-01得票数 1
回答已采纳
1回答
我有以下结构的RDD：rdd2=my_rdd.map(lambda f: f.text.split()) print(x)
它给了我预期的输出然而，当我尝试
浏览 7提问于2021-03-27得票数 0
1回答
我想知道我是否可以在Google Dataflow平台上直接缓存数据集(就像在Spark中缓存RDDs一样)。如果没有这样的功能，Dataflow如何在应用程序中挑选热门数据集，特别是当您有多个热门数据集，并且您希望根据数据集的重要性来确定缓存的优先级时？
浏览 1提问于2017-09-02得票数 2
2回答
我试图编写一个函数，它应该根据输入返回不同的对。我已经重写了Scala中的"+ -/ *“以供我的特定用途。每个实现( +，-，*，/)都有三个基于输入的实现。现在，我有了一个解析器，它从输入(如: RDD+1 )读取表达式，解析它并创建后缀，以使计算更容易，比如: RDD1+，然后我想使用实现的+进行计算。在的帮助下，我试图
浏览 3提问于2015-07-02得票数 0
回答已采纳
1回答
我想知道Apache是如何实现洗牌阶段的。它是否使用与MapReduce相同的技术？rddA.map1.groupByKey();rddY = rddB.map4.map5.saveAsTextFile();
它是否执行然后读取中间文件2次，一次用于map2 map3分支，另一次用于map4 map5，而无需再次计算rddB，即使我们没有在rddB上执行隐式缓存？
浏览 1提问于2015-04-21得票数 0
回答已采纳
1回答
假设我有3个简单的SparkML模型，它们将使用相同的DataFrame作为输入，但完全独立于彼此(无论是在运行的序列中还是在所使用的数据列中)。我想到的第一件事就是创建一个管道数组，其中包含三个阶段数组中的模型，并运行总体拟合/转换来获得完整的预测等等。但是，我的理解是，因为我们将这些模型作为一个序列堆叠在一个管道<
浏览 3提问于2021-01-04得票数 0
回答已采纳
如何在Spark streaming中获取当前批量时间戳(DStream)？在执行过程中，我需要当前的时间戳来验证输入数据中的时间戳。如果我与当前时间进行比较，那么每次RDD转换执行时，时间戳可能会有所不同。
有没有办法获得时间戳，当特定的Spark s
浏览 4提问于2015-12-23得票数 4
我的假设是，星火数据帧是由RDDs构建的。然而，我最近了解到情况并非如此，很好地解释了它们不是这样的。
那么，将RDD转换为DataFrame并再次返回的开销是多少？它是可以忽略不计的还是重要的？在我的应用程序中，我通过将文本文件读取到RDD中来创建DataFrame，然后使用返回Row()对象的映射函数定制编码每一行。我不应该这么做吗？有没有更
浏览 0提问于2019-04-27得票数 0
回答已采纳
2回答
我在Spark中有一个RDD，其中的对象基于case类：我想使用Spark的ML管道，所以我将其转换为Spark数据帧作为管道的一部分，我希望将其中一列转换为其条目为向量的列。由于我希望该向量的长度随模型的不同而变化，因此应该将其作为特征<e
浏览 2提问于2016-03-18得票数 6
2回答
转换基于现有的RDD创建新的RDD。基本上，RDDs是不可变的，而Spark中的所有转换都是惰性的。RDDs中的数据在执行操作之前不会被处理，但是如果不处理数据，如何创建新的RDDs？例如，在filter操作中，如何在不实际将RDD加载到内存并进行处理<em
浏览 2提问于2016-09-03得票数 4
回答已采纳
2回答
在Spark中创建二进制直方图
、、、、
假设我有一个包含以下两列的dataframe (df) (Pandas)或RDD (Spark)：12345.0 10 在Pandas中，我可以很容易地创建不同bin长度的二进制直方图。例如，要创建一个超过1小时的直方图，我执行以下操作：
df = df[ ['timestamp', 'data'] ].s
浏览 1提问于2015-12-29得票数 1
2回答
我想在MLlib中使用管道的实现。以前，我有一个RDD文件并将其传递给模型创建，但是现在要使用管道，应该有要传递给管道的LabeledDocument序列。我有我的RDD，它创建如下：val parsedData = data.map { line =>
val partsline.spli
浏览 1提问于2015-06-19得票数 1
回答已采纳
我有一个RDD，rdd = sc.Textfile(file.txt)，我调用rdd.persist()，然后重新分配它：rdd = rdd.filter(lambda x: 'hi' in x)。现在是将sc.Textfile(file.txt)持久化在内存中，还是将rdd.filter(lambda x: 'hi' in x)持久化？
浏览 0提问于2018-02-20得票数 0
1回答
这两种连接两种数据格式的方法有什么区别？
在数据文件中使用"createOrReplaceTempView“，并使用sparkSession.sql()。对两个dataframes使用dataframe.alias()，然后使用join()方法
浏览 0提问于2018-12-17得票数 2
回答已采纳
1回答
对星火中的action and transformation有一些疑问。 我使用的是过去几个月的spark API。(Learned) Spark api有一个强大的功能，在采取任何操作将最终转换后的数据存储到某个地方之前，它不会将任何数据加载到内存中。这是正确的理解吗？更精确的定义： Spark将使用applied operation, source R
浏览 18提问于2020-04-12得票数 0
1回答
我正在尝试使用spark从亚马逊s3获取一个文件(以DataFrame或RDD的形式)，执行一些简单的转换，然后将文件发送到DynamoDB上的一个表中。在阅读了其他一些论坛帖子后，我开始了解到读/写DynamoDB需要使用hadoopRDD -这与spark中的RDD不同-也不同于我检索s3文件的方式。我该如何将Da
浏览 0提问于2016-05-26得票数 7
1回答
我可以在Hadoop中使用它们，方法是创建一个作业对象，然后将所需的Mapper和Reducer类设置如下： j.setMapperClass(CustomMapper.class);我如何在使用Java的星火中实现同样的目标？我创建了一个java对象，如下
浏览 4提问于2020-02-23得票数 0
我对apache ignite和spark都是新手。在scala中，有没有人可以通过示例将ignite rdd转换为spark rdd？更新-使用案例:我将收到hbase表的数据帧..我将执行一些逻辑来构建报告，将其保存到ignite rdd中。将为每个表更新相同的ignite rdd ...一旦所有的表都被执行，最终的igni
浏览 3提问于2016-09-26得票数 0
我正在思考如何使用spark streaming创建一个类似于storm的多级(基于图形的拓扑)流应用程序。在风暴中，我们有了喷嘴，螺栓的概念，我们可以将这些概念链接起来形成一个图，不同种类的螺栓链接在一起来执行不同的任务。螺栓/喷嘴可以独立地执行它们的工作，并将结果进一步传递到下游。如何使用spark streaming实现类似的功能
浏览 1提问于2020-08-10得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
