在Spark中划分正常的RDD
在Spark中，划分正常的RDD是指将一个RDD划分为多个较小的分区，以便在集群中并行处理数据。RDD（弹性分布式数据集）是Spark中的基本数据结构，它代表了一个可并行操作的不可变分布式集合。
划分RDD的主要目的是提高数据处理的效率和性能。通过将数据划分为多个分区，Spark可以将这些分区分配给集群中的不同节点进行并行处理，从而加快数据处理的速度。此外，划分RDD还可以帮助优化数据的存储和传输，减少网络通信的开销。
在Spark中，可以使用以下方法来划分正常的RDD：
- 基于数据源：可以根据数据源的特点将数据划分为多个分区。例如，如果数据源是一个文本文件，可以按照文件的行数或文件的大小将数据划分为多个分区。
- 基于键值对：如果RDD包含键值对类型的数据，可以根据键的哈希值将数据划分为多个分区。这样可以确保具有相同键的数据被分配到同一个分区中，方便后续的聚合操作。
- 手动指定分区数：可以通过调用RDD的
repartition或coalesce方法手动指定分区数。repartition方法会将RDD的数据重新分区，而coalesce方法可以在不进行数据重分区的情况下减少分区数。
划分正常的RDD可以在以下场景中发挥作用：
- 数据并行处理：通过将数据划分为多个分区，可以实现数据的并行处理，提高处理速度和效率。
- 分布式计算：划分正常的RDD可以将计算任务分配给集群中的多个节点进行并行计算，加快计算速度。
- 数据聚合：通过将具有相同键的数据划分到同一个分区，可以方便进行后续的聚合操作，如求和、求平均等。
腾讯云提供了一系列与Spark相关的产品和服务，包括云数据处理服务、云大数据计算服务等。您可以通过访问腾讯云官方网站（https://cloud.tencent.com/）了解更多相关信息。
相关·内容
我试图计算负样本的数量，如下所示：
val numNegatives = dataSet.filter(col("label") < 0.5).count
但我得到的大小超过了Integer.MAX_VALUE错误：
java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE
at sun.nio.ch.FileChannelImpl.map(FileChannelImpl.java:869)
at org.apache.spark.storage.DiskStore$$anonfu
浏览 0提问于2018-04-13得票数 0
回答已采纳
在的例子中
// Perform word count.
val wordCounts = (tableData
.map(entry => convertToTuple(entry._2))
.reduceByKey(_ + _))
// Write data back into a new BigQuery table.
// IndirectBigQueryOutputFormat discards keys, so set key to null.
(wordCounts
.map(pair => (null, convertToJson(pai
浏览 2提问于2017-10-07得票数 1
我正在尝试使用Spark (在EMR上是2.1 )处理~500M的gz文件，我没有办法改变格式或将它们分割成更小的尺寸。
其中一个执行程序失败，出现以下问题：
java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE
at sun.nio.ch.FileChannelImpl.map(FileChannelImpl.java:869)
at org.apache.spark.storage.DiskStore$$anonfun$getBytes$2.apply(DiskStore.scala:103)
浏览 10提问于2017-05-21得票数 2
1回答
我是Scala和RDD概念的新手。在Spark中使用读取kafka的消息，并尝试在业务工作后提交。但我错了。
注:在并行工作中使用重新分区
如何从流APi中读取偏移量并将其提交给Kafka？
scalaVersion := "2.11.8“val sparkVersion = "2.2.0”val connectorVersion = "2.0.7“Valkafka_stream_version= "1.6.3”
码
val ssc = new StreamingContext(spark.sparkContext, Seconds(2))
浏览 0提问于2018-03-18得票数 0
1回答
我正在尝试将PostgreSQL 9.6中的一个表读入RDD中的Spark2.1.1，我在Scala中有如下代码。
import org.apache.spark.rdd.JdbcRDD
import java.sql.DriverManager
import org.apache.spark.SparkContext
val sc = SparkContext.getOrCreate()
val rdd = new org.apache.spark.rdd.JdbcRDD(
sc,
() => {DriverManager.getConnection(
&#
浏览 0提问于2017-06-16得票数 2
回答已采纳
1回答
当我执行以下命令时：
scala> val rdd = sc.parallelize(List((1,2),(3,4),(3,6)),4).partitionBy(new HashPartitioner(10)).persist()
rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[10] at partitionBy at <console>:22
scala> rdd.partitions.size
res9: Int = 10
scala> rdd.partitioner.isDefined
浏览 1提问于2015-12-28得票数 20
回答已采纳
我需要parallelize我的数据集，但是，当我试图使用重新分区的数据实例化一个RowMatrix时，会发生type mismatch错误。
下面是读取和重新分区数据的代码：
val data = sc.textFile("data.txt.gz").flatMap(r => r.split(' ') match {
case Array(doc, word, count) => Some((doc.toInt, (word.toInt - 1, count.toDouble)))
case _ => None
}).groupByK
浏览 3提问于2017-04-20得票数 0
回答已采纳
1回答
我正在从HDFS获取数据，并将其存储在Spark RDD中。Spark根据HDFS块的数量创建分区的数量。这会导致大量的空分区，这些分区也会在管道过程中被处理。为了消除这种开销，我想从RDD中过滤出所有空分区。我知道合并和重新分区，但不能保证所有的空分区都会被删除。
有没有其他的办法来解决这个问题？
浏览 2提问于2015-10-22得票数 10
我有一个pyspark，它可以收集成一个元组列表，如下所示：
rdds = self.sc.parallelize([(("good", "spark"), 1), (("sood", "hpark"), 1), (("god", "spak"), 1),
(("food", "spark"), 1), (("fggood", "ssspark"), 1), (("
浏览 2提问于2020-06-17得票数 2
回答已采纳
我在执行星火应用程序时遇到了问题。
源代码：
// Read table From HDFS
val productInformation = spark.table("temp.temp_table1")
val dict = spark.table("temp.temp_table2")
// Custom UDF
val countPositiveSimilarity = udf[Long, Seq[String], Seq[String]]((a, b) =>
dict.filter(
(($"first"
浏览 3提问于2017-11-04得票数 10
回答已采纳
1回答
多亏了SparkR的Spark SQL模块，我尝试测试Window function。我使用Spark1.6，并尝试在两种不同的部署模式(local和yarn-client)中复制提供的示例。
set.seed(1)
hc <- sparkRHive.init(sc)
sdf <- createDataFrame(hc, data.frame(x=1:12, y=1:3, z=rnorm(12)))
registerTempTable(sdf, "sdf")
query <- sql(hc, "SELECT x, y, z, LAG(z) OVER
浏览 0提问于2016-01-22得票数 2
1回答
Spark包含一个集合，每个元素代表一个请求。
Scala函数将传递给RDD，对于每个RDD元素，该函数将创建一个修改的请求。
对于每个集合元素\请求，需要引用一个查找表。参考表的最大大小为200行。
如何实现性能和可伸缩性，如何对查找表(在函数中使用)进行建模？
火花广播变量。
分开星火RDD。
Scala不变集合。
也许我还没有考虑过另一种选择。
谢谢
浏览 4提问于2016-01-28得票数 1
回答已采纳
我试图使用TitanDB (从卡夫卡队列中收集消息)向SparkStreaming添加元素。但这似乎比预期的要难。这里是泰坦连接的定义：
val confPath: String = "titan-cassandra-es-spark.properties"
val conn: TitanModule = new TitanModule(confPath)
Titan模块是一个配置Serializable连接的TitanDB类：
...
val configurationFilePath: String = confFilePath
val configuration = n
浏览 2提问于2016-05-30得票数 1
回答已采纳
我记得在使用RDD时，如果一个键值RDD (rdd1)有一个已知的分区，那么使用不同的、不分区的键值RDD (rdd2)执行一个连接就会提高性能。这是因为1)只有rdd2的数据需要在网络上传输，2) rdd2的每个元素只需要传输到一个节点，而不是所有节点，方法是将rdd1的密钥划分到rdd2的密钥上。
我正在学习与DataFrames的洗牌排序合并连接。我正在读的这本书(学习火花，第二版)中的例子是为了加入两个基于DataFrames列的user_id。该示例试图演示如何从联接操作中消除Exchange阶段，因此，在连接之前，两个DataFrames将被加入的列放入相同数量的桶中。
我的问题是
浏览 2提问于2020-08-06得票数 0
回答已采纳
我正在使用RandomForestCLassifier构建一个模型。这是我的密码
conf = SparkConf()
conf.setAppName('spark-nltk')
sc = SparkContext(conf=conf)
sqlContext = SQLContext(sc)
m=sc.textFile("Question_Type_Classification_testing_purpose/data/TREC_10.txt").map(lambda s: s.split(" ",1))
df= m.toDF()
创建的data
浏览 6提问于2017-04-01得票数 3
回答已采纳
2回答
我有一个DSE集群，集群中的每个节点都在运行spark和Cassandra。
当我从Cassandra加载数据到spark rdd并在Rdd上执行一些操作时，我知道数据将分布到多个节点中。在我的例子中，我想把这些rdds从每个节点直接写到它的本地Cassandra dB表中，有没有办法做到这一点。
如果我执行普通的rdd收集，来自spark节点的所有数据将被合并，并返回到带有驱动程序的节点。我不希望发生这种情况，因为从节点返回到驱动程序节点的数据可能需要很长时间，我希望数据直接保存到本地节点，以避免数据跨spark节点移动。
浏览 3提问于2020-10-20得票数 1
1回答
我的卡夫卡主题中有数以百万计的信息，不幸的是，无论我得到什么时间限制，星火流似乎都在排队等待所有的消息。我已经设置了16核和64 of的独立服务器，我给了我的驱动程序12G和执行器12G内存。
16/09/14 19:27:45 WARN TaskSetManager: Lost task 4.0 in stage 1.0 (TID 7, 10.206.41.172): java.lang.OutOfMemoryError: GC overhead limit exceeded
at com.twitter.chill.Tuple4Serializer.read(TupleSe
浏览 1提问于2016-09-14得票数 1
回答已采纳
2回答
如何在使用orderBy后确定分区的数量？我一直认为生成的数据true具有spark.sql.shuffle.partitions，但这似乎不正确：
val df = (1 to 10000).map(i => ("a",i)).toDF("n","i").repartition(10).cache
df.orderBy($"i").rdd.getNumPartitions // = 200 (=spark.sql.shuffle.partitions)
df.orderBy($"n").rdd.getNu
浏览 1提问于2018-12-14得票数 5
回答已采纳
1回答
我想用spark编写一个自定义分区程序，并且我正在编写java。
但是，我注意到，javaRDD类(或Dataset类)没有像scala中那样的partitionBy(分区)方法。只有javaPairRDD这样做。如果没有此方法，我应该如何划分RDD或数据集？
浏览 0提问于2017-11-26得票数 2
回答已采纳
我使用Spark将2100万条记录插入到Cassandra表中。spark作业大约需要一个小时，并成功插入所有记录，但在最后阶段(62/63)挂起。
我的Spark属性：
spark.driver.memory 10g
spark.executor.cores 1
spark.executor.instances 40
spark.executor.memory 4g
将记录插入到Cassandra并在最后阶段挂起的代码行-
raw_data_final.rdd.saveToCassandra("marketing_dev1", "offer_detail_11"
浏览 1提问于2018-06-23得票数 0
我目前正在spark上运行一个流媒体应用程序，它使用新的直接方法(没有接收器)消费一个简单的kafka主题。由于我们在Kafka集群上遇到了一些问题，即使在Kafka上恢复正常后，流媒体应用程序也遇到了一些问题:一些作业正在处理几乎空的rdd，而另一些作业正在处理两倍大小的rdd。平均输入速率仍然与以前相同，但处理时间因此不规律。
我试着重启应用程序，但没有什么不同。我在Kafka和Spark两边都没有发现任何堆栈或线索。
有人能告诉我如何解决这类问题吗？或者至少到哪里去看一看，以了解发生了什么？
浏览 0提问于2016-05-18得票数 0
1回答
我正在亚马逊的EMR上运行一个Spark作业，在客户端模式下，使用吡火花来处理两个大小为200 GB的输入文件中的数据。
作业将数据连接在一起(使用reduceByKey)，执行一些映射和过滤器，并以Parquet格式将其保存到S3中。当作业使用Dataframes进行保存时，所有实际的转换和操作都是在RDDs上执行的。
注意，我已经包含了我的当前配置和值的详细概述，我已经在“故障”部分之后对这些配置和值进行了实验。
代码
与我们所看到的失败相关的代码发生在reduceByKey步骤中。我已经包含了几行上下文，以显示一个先前的map函数和在RDD上实际触发reduceByKey的保存操作：
浏览 1提问于2017-03-28得票数 5
回答已采纳
我使用的是Spark3.0.1，用户提供了Hadoop 3.2.0和运行在Kubernetes上的Scala2.12.10。
当读取压缩为snappy的拼花文件时，一切正常，但是当我尝试读取压缩为zstd的拼板文件时，有几个任务在以下错误下失败：
java.io.IOException: Decompression error: Version not supported
at com.github.luben.zstd.ZstdInputStream.readInternal(ZstdInputStream.java:164)
at com.github.luben.zstd.ZstdInp
浏览 4提问于2020-11-17得票数 6
1回答
我们正在构建简单的流应用程序，它使用HBase RDD与传入的DStream连接。样本代码：
val indexState = sc.newAPIHadoopRDD(
conf,
classOf[TableInputFormat],
classOf[ImmutableBytesWritable],
classOf[Result]).map { case (rowkey, v) => //some logic}
val result = dStream.transform { rdd =>
rdd.leftOuterJoin(indexState)
}
它可
浏览 1提问于2017-02-16得票数 1
回答已采纳
我得到了几个Avro文件，它们是从java类A的对象生成的。使用spark-shell (Spark 1.4.0)，我可以使用spark-avro (2.0.1版)将这些文件读取到数据帧中。
val df = sqlContext.read.avro("file.avro")
只要spark shell不知道类A，就可以正常工作。如果我向shell添加一个jar，并发出相同的命令，我会得到以下异常：
A cannot be cast to org.apache.avro.generic.IndexedRecord
at org.apache.avro.generic.G
浏览 1提问于2015-09-10得票数 1
我有一个小的Scala代码，它可以在Spark-Shell上正常工作，但不能在带有Scala插件的Eclipse中工作。我可以使用插件访问hdfs试着写另一个文件，它工作了..
FirstSpark.scala
package bigdata.spark
import org.apache.spark.SparkConf
import java. io. _
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
object FirstSpark {
def main(args: Arra
浏览 8提问于2016-07-20得票数 0
回答已采纳
我试图使用InfoGain第三方()软件包的MRMR功能.但是我的集群是2.0，我得到了这个异常。即使我添加了所有必需的Jar文件来激发类路径。但它仍然不起作用。虽然它在本地机器上正常工作，但不在集群上工作。
例外：
18/03/29 01:16:43 WARN TaskSetManager: Lost task 3.0 in stage 14.0 (TID 47, EUREDWORKER3): java.lang.NoSuchMethodError: breeze.linalg.DenseMatrix$.canMapValues(Lscala/reflect/ClassTag;)Lbreez
浏览 0提问于2018-03-28得票数 0
回答已采纳
法国城市数据(会计、企业、就业.)可以方便地按部门(29=Finistère、33=Gironde等)进行分区，在我的数据集中，这个字段名为codeDepartement。
对于这些城市，我有许多企业、会计、税收等方面的参考资料，它们是按codeCommune在这些部门分区内分类的：
enterprises = enterprises.repartition(col("codeDepartement")).sortWithinPartition("codeCommune")
然后，我想加入他们分析的时候来了：
I为master.配置了local[12]
我想研
浏览 1提问于2021-05-06得票数 0
该错误可以在火花壳中再现。基本上，我定义了一个具有生成RDD的方法的类，然后在RDD上执行一个映射操作，这将生成序列化错误。
如果我没有方法，只有执行方法步骤的语句，那么一切都正常。
代码在这里，可以在shell中运行，我定义了一个类，然后实例化这个类。
首先是进口
import java.nio.file.{Files}
import java.io._
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import scala.io.Source
import scala.collection.muta
浏览 1提问于2015-08-18得票数 4
我让连接正常工作，但两天后，从表中提取数据出现了问题。星火的配置没什么改变..。
简单步骤1-从HAWQ中的简单表打印模式，我可以创建一个SQLContext DataFrame并连接到HAWQ：
df = sqlContext.read.format('jdbc').options(url=db_url, dbtable=db_table).load()
df.printSchema()
其中的指纹：
root
|-- product_no: integer (nullable = true)
|-- name: string (nullable = true)
|--
浏览 6提问于2015-09-24得票数 1
回答已采纳
我正在为一个星火方法编写单元测试，它以多个数据帧作为输入参数，并返回一个数据帧。火花方法的代码如下所示：
class processor {
def process(df1: DataFrame, df2: DataFrame): DataFrame = {
// process and return resulting data frame
}
}
相应单元测试的现有代码如下：
import com.holdenkarau.spark.testing.DataFrameSuiteBase
import org.apache.spark.sql.DataFrame
浏览 5提问于2017-12-19得票数 0
回答已采纳
2回答
我正在使用Spark的Python并运行Spark0.8。
我正在存储大量的浮点向量的RDD，我需要对整个集合执行一个向量的计算。
RDD中的片和分区之间有什么区别吗？
当我创建RDD时，我将它作为一个参数传递给它，这将导致它将RDD存储为100片，并在执行计算时创建100个任务。我想知道，通过使系统能够更有效地处理数据，对数据进行分区是否会提高切片以外的性能(即，对分区执行操作与仅对切片的RDD中的每个元素执行操作之间是否有区别)。
例如，这两段代码之间有什么显著的区别吗？
rdd = sc.textFile(demo.txt, 100)
vs
rdd = sc.textFile(demo.
浏览 3提问于2014-06-17得票数 12
回答已采纳
我正在用"com.holdenkarau.spark-testing-base“和scalatest来测试一个星火流应用程序。
import com.holdenkarau.spark.testing.StreamingSuiteBase
import org.apache.spark.rdd.RDD
import org.scalatest.{ BeforeAndAfter, FunSuite }
class Test extends FunSuite with BeforeAndAfter with StreamingSuiteBase {
var delim: Strin
浏览 7提问于2017-02-07得票数 11
回答已采纳
2回答
因此，我试图创建一个非常简单的火花笔记本使用Azure数据库，并希望利用一个简单的RDD地图调用。
这只是为了乱搞，所以这个例子有点做作，但是我无法在RDD映射调用中得到一个值，除非它是一个静态常量值。
我试过使用广播变量
下面是一个使用int的简单示例，我广播了它，然后尝试并在RDD映射中使用。
val sparkContext = spark.sparkContext
val sqlContext = spark.sqlContext
import sqlContext.implicits._
val multiplier = 3
val multiplierBroadcast = sp
浏览 0提问于2018-04-26得票数 1
回答已采纳
当我连接到一个运动流时，我得到了以下错误消息。
java.lang.NoSuchMethodError: org.apache.spark.storage.BlockManager.get(Lorg/apache/spark/storage/BlockId;)Lscala/Option;
at org.apache.spark.streaming.kinesis.KinesisBackedBlockRDD.getBlockFromBlockManager$1(KinesisBackedBlockRDD.scala:104)
我的火花流代码是，
sc = SparkContext(app
浏览 2提问于2017-04-03得票数 0
回答已采纳
我尝试在我的pc上安装Spark Release 2.4.0，它的系统是win7_x64。
但是，当我尝试运行简单的代码来检查spark是否可以工作时:代码：
import os
from pyspark import SparkConf, SparkContext
conf = SparkConf().setMaster('local[*]').setAppName('word_count')
sc = SparkContext(conf=conf)
d = ['a b c d', 'b c d e', 'c d e f
浏览 2提问于2018-11-19得票数 2
这是错误日志-
FileBasedWriteAheadLogSegment(s3n://*****/checkpoint/receivedData/20/log-1439298698600-1439298758600,13678,5069)：无法读取预先写入日志记录中的数据，org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1(WriteAheadLogB
浏览 2提问于2015-08-12得票数 1
我在Spark中读取了许多ORC文件，并对其进行了处理，这些文件基本上都是Hive分区。大多数情况下，处理过程都很顺利，但对于少数文件，我得到了以下异常，不知道为什么？这些文件在使用配置单元查询的配置单元中工作得很好。
DataFrame df = hiveContext.read().format("orc").load("/path/in/hdfs");
java.lang.NullPointerException
at org.apache.spark.sql.hive.HiveInspectors$class.unwrapperFor(Hive
浏览 1提问于2015-09-08得票数 1
在我的星火应用程序中，我以如下方式从一个Kafka主题创建了一个DStream：
KafkaUtils
.createDirectStream[String, Array[Byte], StringDecoder, DefaultDecoder, (String, Array[Byte])](
streamingContext,
kafkaParams,
offset.get,
{ message: MessageAndMetadata[String, Array[Byte]] => (message.key(), message.message()
浏览 3提问于2017-05-30得票数 1
回答已采纳
在我有很多工作要处理但集群很小的情况下，我可以让Apache spark以某种方式为我平滑工作，这样集群就不会因为内存不足而爆炸吗？ 在这里，大量工作意味着比可用内存总量更多。 在这种情况下，这将是有价值的，例如，在运行我的调度批处理作业时，由于某种原因，我们有不寻常的数据量要处理。这种情况应该很少见，但仍有可能发生。我不想仅仅因为这些罕见的情况来评估我的集群。 对于这些罕见的情况，我如何才能让Apache Spark平滑工作负载，以使其适合集群并避免内存不足？在这里，处理这种平滑的执行时间较长是可以接受的。 如果正常情况下可用RAM的数量远远少于要处理的数据量，那么当然应该增加集群以适应这
浏览 16提问于2020-04-13得票数 0
回答已采纳
2回答
我正在读取HDFS中的一个文件，我希望将它保存在另一个存储库中，并使用DataFrame结构。我的数据的一个例子：
05-09-2020 22:10 10，jony，abcd，usr.admin.local.teste
我想将这个数据结构发送到另一个存储库：
05-09-2020 22:10:10，jony，abcd，
当我在scala中编写这段代码时，一切都很好，一切正常：
val read = sc.textFile("hdfs://.../teste.csv")
val select = read.map(_.split(",")).map{x
浏览 2提问于2020-09-05得票数 0
1回答
我尝试在循环的每个迭代中将一个条目附加到现有的RDD。到目前为止，我的代码是：
var newY = sc.emptyRDD[MatrixEntry]
for (j <- 0 until 8000) {
var arrTmp = Array(MatrixEntry(j, j, 1))
var rddTmp = sc.parallelize(arrTmp)
newY = newY.union(rddTmp)
}
在进行这8000次迭代时，当我尝试从该RDD中获取(10)时，我得到了一个错误，但如果我尝试使用较小的数字，一切都是正常的。error Exception in thr
浏览 3提问于2018-01-29得票数 0
目前使用的是GCP和Dataproc，我对apache、pyspark和debian都是新手。因此，我试图在dataproc集群(Debian )中复制一个在本地机器(W10，VS Code，Spark3.3.1)中完美运行的火花作业。通过JDBC驱动程序从Server摄取数据到Spark数据。
当我在这个Debian中尝试时，SparkSession.read()正常工作，但dataframe.show()却不能工作。
Debian配置: Debian 10与Hadoop3.2和Spark3.1.3。
JDBC驱动程序:MSSQLJDBC-11.2.1.jre8.jar
Java : op
浏览 11提问于2022-11-17得票数 1
回答已采纳
1回答
在Spark应用程序(v2.3.3)中，我想使用scala测试中的Wiremock。我使用以下依赖项：
"org.apache.spark" %% "spark-sql" % "2.3.3" % "provided"
"org.apache.spark" %% "spark-mllib" % "2.3.3" % "provided"
"com.github.tomakehurst" % "wiremock" % "2.25.1&
浏览 3提问于2019-11-14得票数 2
我使用Scala (2.10.4版本)和Spark -我已经迁移到Spark 1.0.1。版本，并注意到我的一个脚本现在不能正常工作。它以以下方式使用MLlib库中的k-means方法。
假设我有一个名为KMeansModel的clusters对象
scala> clusters.toString
res8: String = org.apache.spark.mllib.clustering.KMeansModel@689eab53
下面是我的方法，以及在编译时收到的一个错误：
scala> def clustersSize(normData: RDD[Array[Double]
浏览 2提问于2014-07-25得票数 0
回答已采纳
2回答
试图理解Hive分区是如何与Spark分区相关联的，最后提出了一个关于联接的问题。
我有两个外部Hive表；这两个表都由S3桶支持，并由date分区；因此，在每个桶中都有带有名称格式date=<yyyy-MM-dd>/<filename>的键。
问题1:
如果我把这些数据读入星火中：
val table1 = spark.table("table1").as[Table1Row]
val table2 = spark.table("table2").as[Table2Row]
那么，生成的数据集将分别有多少个分区？分区等于S3中的对象数？
浏览 4提问于2018-04-25得票数 5
用Scala编写的快速作业异常中止: java.lang.ClassCastException: com.$Class1不能强制转换为com.$Class1。 Class1是存储在RDD中的自定义类。有趣的是，这个错误是在强制转换同一个类时抛出的。到目前为止，还没有发现任何模式。 在工作中，我们从hbase获取数据，使用Dataframe使用分析元数据丰富数据，并将其推送到SnappyData中的表中。我们使用的是Snappydata 1.2.0.1。 不确定为什么会发生这种情况。 下面是堆栈跟踪:由于阶段故障而中止的作业:阶段42.0中的任务76失败了4次，最近一次失败:阶段42.0中丢失
浏览 14提问于2019-01-18得票数 0
回答已采纳
我正在尝试使用星火协同过滤实现推荐系统。
首先，我准备模型并保存到磁盘：
MatrixFactorizationModel model = trainModel(inputDataRdd);
model.save(jsc.sc(), "/op/tc/model/");
当我使用单独的进程加载模型时，程序会失败，例外情况如下：
代码：
static JavaSparkContext jsc ;
private static Options options;
static{
SparkConf conf = new SparkConf().
浏览 1提问于2016-08-17得票数 1
回答已采纳
1回答
在尝试朴素贝叶斯示例时，我在Ubuntu上的Spark1.4上遇到了这个问题。我见过有类似问题的帖子，修复的是jar不匹配(通过Maven)，但在这种情况下，有问题的类是用Spark打包的，所以我不确定如何继续。
scala> val model = NaiveBayes.train(training, lambda = 1.0, modelType = "multinomial")
<console>:46: error: type mismatch;
found : org.apache.spark.rdd.org.apache.spark.rdd.
浏览 2提问于2015-07-11得票数 0
我在投机模式下运行Spark作业。我有大约500个任务和大约500个文件的1 GB gz压缩。我一直在每个作业中，对于1-2个任务，附加的错误在之后重新运行几十次(阻止作业完成)。
org.apache.spark.shuffle.MetadataFetchFailedException:缺少随机播放0的输出位置
你知道这个问题的意义是什么吗?如何克服它？
org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle 0
at org.apache.spar
浏览 440提问于2015-03-06得票数 103
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
