“‘RDD”对象没有属性“sparkSession”
"RDD"是指弹性分布式数据集（Resilient Distributed Dataset），是Apache Spark中的基本数据结构之一。RDD是一个可分区、可并行计算的数据集合，可以在集群中进行分布式处理。
RDD对象没有属性"sparkSession"是因为"sparkSession"是Spark 2.0版本引入的概念，而RDD是在Spark 1.x版本中引入的。在Spark 1.x版本中，RDD是通过SparkContext进行操作的，而不是通过SparkSession。
RDD的优势包括：
- 分布式计算：RDD可以在集群中进行并行计算，充分利用集群资源，提高计算效率。
- 容错性：RDD具有弹性，可以自动恢复计算中的错误，保证计算的可靠性。
- 内存计算：RDD可以将数据存储在内存中，加快数据访问速度，提高计算性能。
- 数据流式处理：RDD支持流式处理，可以实时处理数据流，适用于实时分析和处理大规模数据。
RDD的应用场景包括：
- 大数据处理：RDD适用于大规模数据的处理和分析，可以进行复杂的数据转换和计算。
- 机器学习：RDD可以作为机器学习算法的输入数据集，支持分布式的机器学习计算。
- 图计算：RDD可以用于图计算算法，如社交网络分析、推荐系统等。
- 实时分析：RDD支持流式处理，可以实时处理数据流，适用于实时分析和监控。
腾讯云提供的相关产品和产品介绍链接地址如下：
- 腾讯云Spark：https://cloud.tencent.com/product/spark
- 腾讯云大数据计算服务：https://cloud.tencent.com/product/dc
- 腾讯云机器学习平台：https://cloud.tencent.com/product/tiia
- 腾讯云流计算Oceanus：https://cloud.tencent.com/product/oceanus
请注意，以上答案仅供参考，具体的产品和链接地址可能会有变化，请以腾讯云官方网站为准。
相关·内容
在星火中使用Scala -当我们必须将RDDRow转换为DataFrame时。为什么我们必须将RDDRow转换为元组的case类或RDD，以便使用rdd.toDF() --任何没有为RDDRow提供的特定原因
object RDDParallelize {
def main(args: Array[String]): Unit = {
val spark:SparkSession = SparkSession.builder().master("local[1]")
.appName("learn")
.getOrCreate()
val abc
浏览 8提问于2022-09-11得票数 0
回答已采纳
我开始玩星火2.0.1了。新的Dataset API非常干净，但是我在非常简单的操作中遇到了问题。
也许我错过了什么希望有人能帮上忙。
这些指示
SparkConf conf = new SparkConf().setAppName("myapp").setMaster("local[*]");
SparkSession spark = SparkSession
.builder()
.config(conf)
.getOrCreate();
Dataset<Info> infos = spark.r
浏览 11提问于2016-11-11得票数 4
我正在开发Spark版本: 2.0.1和Cassandra 3.9。我想用CassandraSQLContext从卡桑德拉的一个表中读取数据。然而，Spark 2.0被改变了，并使用了sparkSession。我正在尝试使用sparkSession，我很幸运，以下是我的代码。
你能回顾一下并给出你的建议吗？
def main(args: Array[String], date_filter: String): Unit = {
val conf = new SparkConf(true).set("spark.cassandra.connection.host", &#
浏览 2提问于2016-10-11得票数 0
它们之间的区别是什么？SparkContext, JavaSparkContext, SQLContext和SparkSession？ 是否有任何方法可以使用SparkSession？ 我是否可以使用单个条目完全替换所有上下文SparkSession？ 中的所有函数SQLContext，SparkContext，以及JavaSparkContext也在SparkSession？ 一些函数，比如parallelize在中有不同的行为SparkContext和JavaSparkContext。它们在如何使用SparkSession？ 我如何使用 a 创建以下内容SparkSession？ RDD
浏览 216提问于2017-05-05得票数 39
回答已采纳
1回答
首先，我要告诉大家，我对火花很陌生。
我需要处理一个表中的大量记录，当它通过电子邮件分组时，大约是100万条。我需要对、个人电子邮件、和执行多个逻辑计算，根据逻辑计算更新数据库。
大概我的代码结构就像
//Initial Data Load ...
import sparkSession.implicits._
var tableData = sparkSession.read.jdbc(<JDBC_URL>, <TABLE NAME>, connectionProperties).select("email").where(<CUSTOM CON
浏览 0提问于2017-02-23得票数 0
我在用火花壳来试验星火的HashPartitioner。错误显示如下：
scala> val data = sc.parallelize(List((1, 3), (2, 4), (3, 6), (3, 7)))
data: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[0] at parallelize at <console>:24
scala> val partitionedData = data.partitionBy(new spark.HashPartitioner(2))
<
浏览 1提问于2017-05-24得票数 1
回答已采纳
1回答
在pyspark中可以在python对象上使用parallelize函数吗？我想在一个对象列表上并行运行，使用一个函数修改它们，然后打印这些对象。
def init_spark(appname):
spark = SparkSession.builder.appName(appname).getOrCreate()
sc = spark.sparkContext
return spark,sc
def run_on_configs_spark(object_list):
spark,sc = init_spark(appname="analysis")
p
浏览 11提问于2020-07-31得票数 0
我运行这个脚本时使用的是spark 2.4.3 & python 3.6.10
columns = ["language","users_count"]
data = [("Java", "20000"), ("Python", "100000"), ("Scala", "3000")]
spark = SparkSession.builder.appName('SparkByExamples.com').getOrCreate()
rdd =
浏览 13提问于2021-06-04得票数 1
回答已采纳
package RDD_Programing
import org.apache.spark.sql.SparkSession
object RDD_Programing {
def main (args:Array[String]) : Unit = {
val spark=SparkSession.builder().master(master = "local")
.appName(name = "rdd programing").getOrCreate()
val sc=spark.sparkContext
浏览 1提问于2022-11-02得票数 0
我在Jupyter notebook上使用了pyspark。下面是Spark的设置方法：
import findspark
findspark.init(spark_home='/home/edamame/spark/spark-2.0.0-bin-spark-2.0.0-bin-hadoop2.6-hive', python_path='python2.7')
import pyspark
from pyspark.sql import *
sc = pyspark.sql.SparkSession.builder.master(&
浏览 21提问于2016-09-16得票数 22
回答已采纳
我有以下示例数据:保存为test.json
{
"memberId": "Test0001",
"params": {"field1":"testField","field2":"testField2"}
"withoutValueSample": {}
}
这是我在“星火”上读到的：
val session: SparkSession = SparkSession.builder().getOrCreate()
val inputRdd: R
浏览 1提问于2021-02-24得票数 0
回答已采纳
1回答
用greenplum-connector-apache-spark-scala_2.12-2.1.0 - java.lang.IllegalStateException将星火写入绿梅数据库
格林梅版本: PostgreSQL 9.4.24 (格林梅数据库6.8.1构建提交:xxxxxxx)在x86_64-未知-linux-gnu上，由gcc (Ubuntu7.5.0-3 ubuntu1~18.04) 7.5.0，64位编译于2020年6月16日18:53:13
连接器: greenplum-connector-apache-spark-scala_2.12-2.1.0.jar
火花版本:欢迎使
浏览 9提问于2022-01-14得票数 1
3回答
星火流中的序列化问题
、、、
我对星火如何处理引擎盖下的数据感到非常困惑。例如，当我运行流作业并应用foreachRDD时，其行为取决于变量是从外部范围捕获还是在内部初始化。
val sparkConf = new SparkConf()
dStream.foreachRDD(rdd => {
val spark = SparkSession.builder.config(sparkConf).getOrCreate()
...
})
在这种情况下，我得到了一个例外：
org.apache.spark.streaming.kafka.DirectKafkaInputDStream$DirectKaf
浏览 1提问于2016-09-26得票数 10
2回答
我有一个将class作为参数的方法，如下所示。 val hBaseRDD = spark.sparkContext.newAPIHadoopFile(path,
classOf[org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat[ImmutableBytesWritable, Result]],
classOf[ImmutableBytesWritable],
classOf[Result], conf) 我想写一个方法，它接受参数作为类的类型，然后我可以在它里面调用这行代码。
浏览 14提问于2021-10-22得票数 0
在scala中，代码给出了一个字符串数组，但是在python中，我想要相同的输出，就像scala：代码在Scala:中一样
val spark = SparkSession.builder()
.appName("Test_Parquet")
.master("local[*]")
.getOrCreate()
val sc = spark.sparkContext
val parquetDF = spark.read.csv("data.csv")
parquetDF.coalesc
浏览 2提问于2019-11-25得票数 0
回答已采纳
我有两个如下类型的RDD：
RDD[(int, List[(string, int)]
我想从两个RDD中得到减法集。守则是这样的：
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('xxxxx.com').getOrCreate()
rdd1 = spark.sparkContext.parallelize([(1, [("foo", 101), ("bar", 111)]), (2, [("foobar", 22), (
浏览 5提问于2022-07-12得票数 1
回答已采纳
我正在尝试从spark2中的rdd创建一个数据框。当我尝试将rdd转换为数据帧时，我得到一个错误，说明以下内容
error reading Scala signature of org.apache.spark.sql.package: assertion failed: unsafe symbol Unstable (child of package InterfaceStability) in runtime reflection universe
以下是我的代码
import org.apache.spark.sql.{ SparkSession}
object TestLoad ext
浏览 2提问于2019-07-25得票数 0
我在本地安装了spark 2.2并使用scala
spark会话配置如下
val sparkSession = SparkSession
.builder()
.appName("My application")
.config("es.nodes", "localhost:9200")
.config("es.index.auto.create", true)
.config("spark.streaming.backpressure.initialRate", "1")
浏览 0提问于2018-09-13得票数 0
1回答
我是新来的火花，并试图学习它。我正在尝试使用类从textFile创建数据集。当我执行dataset.show()时，它显示所有空白，列长度显示0。
代码：
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Dataset;
impor
浏览 0提问于2018-09-12得票数 0
回答已采纳
我想使用case类将文本文件转换为dataframe，下面是我的代码。它一直工作到映射拆分，在这里我可以看到使用rdd_metadata_schema.take(1).foreach(arr => print(arr.toList))的值，但是当我检查dataframe时是空的。
case class metadata_schema(
field_name:String,
field_pos:String,
field_dataTyp
浏览 0提问于2018-03-20得票数 2
回答已采纳
1回答
我有一个问题:可以从spark.sql.Row获得自定义对象吗？
当前代码能够将数据推送到Spark行，但我无法将其解压缩。
首先，有一个简单的POJO对象：
public class Event implements Serializable {
private Map<String, Object> fields;
public Event() {
}
public Event(Map<String, Object> fields) {
this.fields = fields;
}
public Map<String,
浏览 8提问于2017-08-29得票数 0
我发现SparkSession没有.parallelize()方法，我们需要再次使用SparkContext来创建RDD吗？如果是这样，在一个程序中同时创建SparkSession和SparkContext是可取的吗？
浏览 0提问于2016-10-06得票数 20
回答已采纳
我有一个Spark Streaming应用程序，可以从Kafka中读取记录中的配置单元表名，例如table1。table2..table3..等。
我想在hive表上执行结构化查询，并将结果流式传输到另一个Kafka主题。
我有这样的
val hqls = rdd
.filter(record => record.value() != null && record.value().trim.length > 0)
.foreach(tableName=> publishData(tableName, sparkSession, kafkaPro
浏览 0提问于2019-02-25得票数 1
我有一个名为“类型参数”的特性，它的一个方法需要能够创建一个空的类型化数据集。
trait MyTrait[T] {
val sparkSession: SparkSession
val spark = sparkSession.session
val sparkContext = spark.sparkContext
def createEmptyDataset(): Dataset[T] = {
import spark.implicits._ // to access .toDS() function
// DOESN&#
浏览 8提问于2017-12-05得票数 1
回答已采纳
我在Scala中有一个现有的代码，并试图用Java编写相同的代码。但也面临着一些问题。
Scala代码
import java.io.{BufferedReader, InputStreamReader}
import java.util.zip.ZipInputStream
import org.apache.spark.SparkContext
import org.apache.spark.input.PortableDataStream
import org.apache.spark.rdd.RDD
def readFile(path: String,minPartitions: I
浏览 4提问于2017-11-02得票数 1
回答已采纳
我有一个有多个独立模块的pyspark程序，每个模块都可以独立地处理数据以满足我的各种需求。但它们也可以链接在一起，以便在管道中处理数据。这些模块中的每一个都构建了一个SparkSession，并各自完美地执行。
但是，当我试图在同一个python进程中连续运行它们时，我会遇到问题。在执行管道中的第二个模块时，second抱怨我试图使用的SparkContext已经停止：
py4j.protocol.Py4JJavaError: An error occurred while calling o149.parquet.
: java.lang.IllegalStateException: Ca
浏览 0提问于2017-01-05得票数 28
回答已采纳
环境: Kafka 10，Spark2.1
我正在尝试存储Kakfa偏移量外部存储。在浏览了Apache Spark网站和一些在线研究之后，我能够编写以下代码。现在得到错误-
"Error:(190, 7) object Assign in package kafka010 cannot be accessed in package org.apache.spark.streaming.kafka010
Assign[String, String](fromOffsets.keys.toList, kafkaParams, fromOffsets)"
我的代码：
impor
浏览 27提问于2018-01-03得票数 0
回答已采纳
2回答
我真的快绝望了。
我尝试做的是，捕获一个流，聚合几秒钟的流数据，然后将其保存为CSV文件。
val sparkSession : SparkSession = SparkSession.builder()
.master("local[*]")
.appName("Streaming")
.config(conf)
//.enableHiveSupport()
.getOrCreate()
所以，我正在捕捉这条小溪
val lines = streamingContext.socketTextStream(HOST, PORT)
val li
浏览 0提问于2017-11-12得票数 0
以下是我的代码，我尝试导入所有内容，但仍然报告错误。 from pyspark import SparkContext
from pyspark.sql import SparkSession
from pyspark.sql import SQLContext
import pyspark
sc = SparkContext(appName="session1")
a = [('Chris', 'Budweiser', 15), ('Chris', 'Becks', 5), ('Chris'
浏览 28提问于2021-09-23得票数 0
1回答
spark 2.0兼容(datastax) cassandra 2.1.13吗？我已经在本地mac上安装了spark 2.1.0，还安装了scala 2.11.x。我正在尝试从安装了datastax 4.8.6 ( spark 1.4和cassandra 2.1.13)的服务器读取cassandra表。
我在spark shell上运行以下代码
spark-shell
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.implicits._
import org.apache.spark.sql.cas
浏览 1提问于2017-03-01得票数 0
1回答
我正在尝试计算两个特性之间的相关性，这些特性正在从两个单独的文本文件中读取，如下所示。
import org.apache.spark.sql.SparkSession
import org.apache.spark.mllib.stat.Statistics
import scala.io.Source
object Corr {
def main() {
val sparkSession = SparkSession.builder
.master("local")
.app
浏览 6提问于2017-02-15得票数 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回答
我感兴趣的是一个很好和清晰的描述星火RDD血统或运算符图的教育目的。我试过.toDebugString()，但是我很难把它打印出来(包括换行等)。这里出什么问题了？
Using Python version 3.6.4 (default, Mar 1 2018 18:36:42)
SparkSession available as 'spark'.
>>> sc
<SparkContext master=local[*] appName=PySparkShell>
>>> rdd = sc.parallelize(range(
浏览 0提问于2018-04-10得票数 1
回答已采纳
1回答
我有以下数据(Alldata)，它具有SQL查询和视图名。
Select_Query|viewname
select v1,v2 from conditions|cond
select w1,w2 from locations|loca
我已经将其拆分并正确地分配给了temptable(alldata)。
val Select_Querydf = spark.sql("select Select_Query,ViewName from alldata")
当我尝试执行查询并从它注册一个tempview或表时，它会显示空指针错误。但是当我注释掉spark.sql stmt时，PR
浏览 3提问于2019-12-09得票数 2
回答已采纳
使用以下单列数据框架， from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('test').getOrCreate()
df = spark.createDataFrame([[1],[2],[3],[4],[5]])
df.show()
+---+
| _1|
+---+
| 1|
| 2|
| 3|
| 4|
| 5|
+---+ 使用rdd直方图函数计算直方图。 df.rdd.histogram(2) 然后我得到一个错误:无法在RDD中生成非number的存储桶。
浏览 31提问于2019-06-10得票数 5
回答已采纳
我正在将测试从JUnit4迁移到JUnit5。代码在JUnit4中运行良好，但在JUnit 5中，它会引发org.apache.spark.SparkException: Task not serializable异常。我试过多种方法，但现在起作用了。在testAlternativeVegetableIdWithDifferentReadCount行中，assertTrue测试失败。当我试图显示Dataset值时，它也会抛出一个错误。我做错什么了？
@ExtendWith(MockitoExtension.class)
public class ExportLogicTest implemen
浏览 5提问于2022-07-11得票数 0
1回答
我有一堂课：
class DataLoader {
def rdd2RddTransform(
ss: SparkSession,
inputRDD: RDD[GenericRecord]): RDD[GenericRecord] = {
inputRDD.asInstanceOf[RDD[TrainingData]]
.map(reformatTrainingData)
}
private def reformatTrainingData: TrainingData => ReFormatedData
浏览 0提问于2019-02-01得票数 4
回答已采纳
2回答
我启用了将日志保存到S3的检查点。如果检查点目录中没有文件，spark streaming可以正常工作，并且我可以看到日志文件出现在检查点目录中。然后我关闭spark streaming并重启它。这一次，我开始为spark session获取NullPointerException。简而言之，如果检查点目录中没有日志文件，spark streaming就可以正常工作。但是，当我在检查点目录中使用日志文件重新启动spark streaming时，我开始在spark会话中获得空指针异常。代码如下：
object asf {
val microBatchInterval = 5
val sp
浏览 5提问于2017-09-13得票数 1
我对PySpark环境很陌生，在尝试用加密模块加密RDD中的数据时遇到了一个错误。下面是代码：
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('encrypt').getOrCreate()
df = spark.read.csv('test.csv', inferSchema = True, header = True)
df.show()
df.printSchema()
from cryptography.fernet import Fernet
ke
浏览 0提问于2017-08-21得票数 8
回答已采纳
当根据我的想法启动spark作业时，我得到以下错误：java.lang.ClassCastException: cannot assign instance of scala.collection.immutable.List$SerializationProxy to field org.apache.spark.rdd.RDD.org$apache$spark$rdd$RDD$$dependencies_ of type scala.collection.Seq in instance of org.apache.spark.rdd.MapPartitionsRDD
我从了解到这可能是一个
浏览 22提问于2016-08-06得票数 0
回答已采纳
我试图理解为什么在这两种方法之间的分区计数没有相同的结果：
val rdd: RDD[Int] = sparkSession.sparkContext.parallelize(0 to 9)
println("S1---> " +rdd.getNumPartitions )
val partitionSizes: Array[Int] = rdd.mapPartitions(iter => Iterator(iter.length)).collect()
partitionSizes.foreach((row: Int) => {
println("
浏览 8提问于2019-10-21得票数 0
回答已采纳
下面是其中一种用法的实际示例：
val a: Either[Int, String] = {
if (true)
Left(42) // return an Int
else
Right("Hello, world") // return a String
}
但是下面的条件不起作用:条件" text“只是用来确定输入文件是文本文件还是parquet文件
val a: Either[org.apache.spark.rdd.RDD[String], org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]]
浏览 2提问于2018-11-17得票数 1
下面是我的代码：
ssc =streamingcontext(sparkcontext,Seconds(time))
spark = sparksession.builder.config(properties).getorcreate()
val Dstream1: ReceiverInputDstream[Document] = ssc.receiverStream(properties) // Dstream1 has Id1 and other fields
val Rdd2 = spark.sql("select Id1,key from hdfs.table"
浏览 4提问于2018-09-01得票数 1
1回答
我到底想做什么？(背景)
我正在试图计算一些数据数据/设置的火花，这是从一个目录读取与.parquet文件有关美国航班在2013年至2015年。更具体地说，我在DataFrameStatFunction中使用DataFrameStatFunction方法，可以访问该方法在Dataset上调用stat方法。
import airportCaseStudy.model.Flight
import org.apache.spark.sql.SparkSession
object CaseStudy {
def main(args: Array[String]): Unit = {
v
浏览 1提问于2020-03-04得票数 1
我尝试运行2个数据流，在第一个数据流中生成Dataframe，将df注册为tmp视图，然后在另一个数据流中使用它，如下所示：
dstream1.foreachRDD { rdd =>
import org.apache.spark.sql._
val spark = SparkSession.builder.config(rdd.sparkContext.getConf).getOrCreate
import spark.implicits._
import spark.sql
val records = rdd.toDF("record")
r
浏览 2提问于2017-03-07得票数 0
2回答
我使用Spark2.1.0、Hadoop2.7.3、Hive2.1.1、Sqoop1.4.6和Ganglia 3.7.2 (从s3加载数据)在EMR ( 5.5.1 )上运行吡火花作业。有多个桶包含输入文件，所以我有一个函数，它使用boto遍历它们，并根据某种模式过滤掉它们。
集群大小：主=> r4.xlarge，Worker => 3 x r4.4xlarge
问题：函数getFilePaths返回一个s3路径列表，该列表直接提供给spark方法。
使用Dataframe
file_list = getFilePaths() # ['s3://some_bucket/lo
浏览 1提问于2018-06-13得票数 1
我发现下面这个简单的示例在Scala REPL (sbt console)中无限期地挂起： import org.apache.spark.sql._
val spark = SparkSession.builder().master("local[*]").getOrCreate()
val sc = spark.sparkContext
val rdd = sc.parallelize(1 to 100000000)
val n = rdd.map(_ + 1).sum 但是，下面的代码运行得很好： import org.apache.spark.sql._
val sp
浏览 20提问于2020-08-22得票数 0
1回答
我是新来的。我想对一个文本文件执行一些机器学习。
from pyspark import Row
from pyspark.context import SparkContext
from pyspark.sql.session import SparkSession
from pyspark import SparkConf
sc = SparkContext
spark = SparkSession.builder.appName("ML").getOrCreate()
train_data = spark.read.text("20ng-train-all-te
浏览 4提问于2018-02-26得票数 2
回答已采纳
2回答
我已经构建了一个使用Apache的推荐系统，它的数据集存储在我的项目文件夹中，现在我需要从HDFS访问这些文件。
如何使用Spark从HDFS读取文件？
我就是这样初始化我的星火会话的：
SparkContext context = new SparkContext(new SparkConf().setAppName("spark-ml").setMaster("local")
.set("fs.default.name", "hdfs://localhost:54310").set("f
浏览 0提问于2019-06-15得票数 2
回答已采纳
1回答
我得到了这个错误，我哪里出错了。请帮帮忙，因为我是spark新手。我如何在RDD上使用mapValues？ 包com.udemyexamples import org.apache.spark.sql.SparkSession
object AverageFriendByAge {
def parseFile(line:String): Unit =
{
val field= line.split(",")
val age=field(2).toInt
val friend=field(3).toInt
(age,friend)
浏览 10提问于2020-01-03得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
