[ spark -cassandra-connector]如何在spark 2.3.1中将scala隐式支持的代码转换为java
在Spark 2.3.1中，将Scala隐式支持的代码转换为Java可以通过以下步骤实现：
- 导入所需的依赖项：
- 导入所需的依赖项：
- 创建SparkSession对象：
- 创建SparkSession对象：
- 创建JavaSparkContext对象：
- 创建JavaSparkContext对象：
- 使用CassandraJavaUtil类执行操作，例如读取数据：
- 使用CassandraJavaUtil类执行操作，例如读取数据：
以上代码示例中，需要将"keyspace"替换为实际的Cassandra keyspace名称，"table"替换为实际的表名称，"column1"和"column2"替换为实际的列名称，"condition"替换为实际的查询条件。
推荐的腾讯云相关产品是TencentDB for Cassandra，它是腾讯云提供的托管式Cassandra数据库服务。您可以通过以下链接了解更多信息： TencentDB for Cassandra
请注意，以上答案仅供参考，具体实现可能因环境和需求而异。
相关·内容
2回答
我有一个Spark (version 1.3.1)应用程序。在其中，我试图将一个Java bean RDD JavaRDD<Message>转换为Dataframe，它有许多不同数据类型的字段(整数、字符串、列表、地图、双数据)。
但是当我在执行我的代码的时候。
messages.foreachRDD(new Function2<JavaRDD<Message>,Time,Void>(){
@Override
public Void call(JavaRDD<Message> arg0, Time
浏览 6提问于2015-06-12得票数 0
回答已采纳
如何解决以下编译错误？
SOApp.scala:7: error: encountered unrecoverable cycle resolving import.
Note: this is often due in part to a class depending on a definition nested within its companion.
If applicable, you may wish to try moving some members into another object.
import spark.implicits._
代码：
object SOAp
浏览 4提问于2020-04-04得票数 3
回答已采纳
当我想使用toDF以便将RDD传递给DataFrame时，我在编译Scala代码时遇到了一些问题。我检查了Spark2.0.0和1.6.2，问题始终是一样的。下面我提供我的POM文件和代码：
POM.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
浏览 1提问于2016-11-15得票数 2
我正在尝试从我的工作站(在Intellij内部)运行一个应用程序，并连接到运行在ec2上的远程星火集群(2.3.1)。我知道这不是一个最好的实践，但如果我能让它为发展而工作，它将使我的生活变得容易得多。
我已经取得了相当大的进展，并且能够在RDDs上运行操作并返回结果，直到我进入使用.zipWithIndex()的步骤，并得到以下异常：
ERROR 2018-07-19 11:16:21,137 o.a.spark.network.shuffle.RetryingBlockFetcher Exception while beginning fetch of 1 outstanding blo
浏览 0提问于2018-07-19得票数 1
回答已采纳
我编写了一个程序，使用火花流将数据插入到启用了hbase的kerberos。在一批中，我遇到了一个失败的任务。错误如下：
java.io.IOException: Login failure for myuser@example.com from keytab ./user.keytab
at org.apache.hadoop.security.UserGroupInformation.loginUserFromKeytabAndReturnUGI(UserGroupInformation.java:1160)
at com.framework.common.HbaseUti
浏览 0提问于2017-05-27得票数 0
回答已采纳
我试图使用spark-cassandra-connector将卡桑德拉行映射为参数化类型。因此，我一直试图使用隐式定义的columnMapper来定义映射：
class Foo[T<:Bar:ClassTag:RowReaderFactory] {
implicit object Mapper extends JavaBeanColumnMapper[T](
Map("id" -> "id",
"timestamp" -> "ts"))
def doSomeStuff(opera
浏览 1提问于2016-07-01得票数 2
回答已采纳
我是Spark和Scala的新手。我试图调用一个函数作为Spark UDF，但我遇到了这个错误，我似乎无法解决。
我知道在Scala中，Array和Seq是不同的。WrappedArray是Seq的一个子类型，并且在WrappedArray和数组之间存在隐式转换，但我不确定为什么在使用自定义函数的情况下不会发生这种转换。
任何能帮助我理解和解决这个问题的建议都是非常感谢的。
这是一段代码片段
def filterMapKeysWithSet(m: Map[Int, Int], a: Array[Int]): Map[Int, Int] = {
val seqToArray = a.toArra
浏览 0提问于2016-10-23得票数 23
回答已采纳
1回答
我将csv文件读取到RDD，并试图将其转换为DataFrame。但是，它克服了错误。
scala> rows.toDF()
<console>:34: error: value toDF is not a member of org.apache.spark.rdd.RDD[Array[String]]
rows.toDF()
scala> rows.take(2)
Array[Array[String]] = Array(Array(1, 0, 3, "Braund, ...
我做错了
浏览 2提问于2015-11-11得票数 2
回答已采纳
2回答
我正在为我的spark应用程序使用spring boot，所有的依赖项都是通过spring来管理的，我正在使用Autowire来添加依赖项。提交给executors的My Function类和Custom类实现了Serializable。
但当我运行它并将任务提交给执行器时，它抛出了异常:一个spring类不是serilazable - AnnotationConfigApplicationContext
Caused by: java.io.NotSerializableException: org.springframework.context.annotation.AnnotationC
浏览 9提问于2018-08-12得票数 0
1回答
我最近开始使用火花。目前，我正在测试一个具有不同顶点和边缘类型的二分图。
根据我在图中所做的研究，为了有不同的边和一些具有属性的边，我需要对这些边进行子类化。
下面是代码的一个片段：
scala> trait VertexProperty
defined trait VertexProperty
scala> case class paperProperty(val paperid: Long, val papername: String, val doi: String, val keywords: String) extends VertexProperty
defined
浏览 0提问于2017-01-16得票数 2
回答已采纳
尝试使用Spark的示例，它们似乎工作得很好，除非需要表达式：
scala> val teenagers = people.where('age >= 10).where('age <= 19).select('name)
<console>:23: error: value >= is not a member of Symbol
val teenagers = people.where('age >= 10).where('age <= 19).select('name)
sca
浏览 0提问于2014-09-09得票数 7
回答已采纳
1回答
我有一个CustomClass特性，我实现了不同类的特性，我使用工厂模式创建了这个类的对象。
import org.apache.spark.sql.Encoders
implicit val encoder = Encoders.kryo[CustomClass ](classOf[CustomClass ])
现在，当我试图创建CustomClass数据时，我得到了以下错误。
dataframe只有一个列值，而且数据是二进制的。
org.apache.spark.sql.AnalysisException: cannot resolve '`id`' given input
浏览 1提问于2021-07-19得票数 0
当我试图运行上面的代码时，我会得到这个错误。
错误:
Exception:
Caused by: java.lang.ClassCastException: com.datastax.driver.core.DefaultResultSetFuture cannot be cast to shade.com.datastax.spark.connector.google.common.util.concurrent.ListenableFuture
at com.datastax.spark.connector.writer.AsyncExecutor.com$da
浏览 1提问于2019-06-26得票数 0
回答已采纳
到目前为止，我已经在本地包目录中使用了这个build.sbt。
name := "spark27_02"
version := "1.0"
scalaVersion := "2.10.4"
sbtVersion := "0.13.7"
libraryDependencies += "org.apache.spark" %% "spark-core" % "1.2.1"
libraryDependencies += "org.apache.spark" %%
浏览 1提问于2015-03-17得票数 3
Getting the below error while saving uuid to postgresql
at org.postgresql.jdbc.PgStatement$BatchResultHandler.handleError(PgStatement.java:2356)
at org.postgresql.core.v3.QueryExecutorImpl$1.handleError(QueryExecutorImpl.java:395)
at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryEx
浏览 12提问于2020-10-22得票数 2
当创建一个列值默认值为NULL的表时，我在Spark(1.6) SQL中获得了下面的错误。示例:将表测试创建为select column_a，NULL作为column_b从test_temp创建；
同样的东西在Hive中工作，并创建数据类型为"void“的列。
我使用空字符串而不是NULL来避免异常和新列获取字符串数据类型。
是否有更好的方法使用spark在hive表中插入空值？
2017-12-26 07:27:59 ERROR StandardImsLogger$:177 - org.apache.hadoop.hive.ql.metadata.HiveException: jav
浏览 3提问于2017-12-27得票数 0
回答已采纳
1回答
我的数据格式是用apache thrift定义的，代码由scrooge生成。我使用拼图将其存储在spark中，非常类似于此中所解释的内容。
我可以很容易地将这些数据读回到Dataframe中，只需执行以下操作：
val df = sqlContext.read.parquet("/path/to/data")
我可以在RDD中读到更多关于体操的内容：
def loadRdd[V <: TBase[_, _]](inputDirectory: String, vClass: Class[V]): RDD[V] = {
implicit val ctagV: Clas
浏览 0提问于2016-02-06得票数 4
1回答
我是星火的新手。您能知道下面的代码有什么问题吗？
val rawData="""USA | E001 | ABC DE | 19850607 | IT | $100
UK | E005 | CHAN CL | 19870512 | OP | $200
USA | E003 | XYZ AB | 19890101 | IT | $250
USA | E002 | XYZ AB | 19890705 | IT | $200"""
val sc = ...
val data= rawData.split("\n")
val r
浏览 2提问于2014-07-10得票数 2
回答已采纳
我对Scala很陌生，并试图掌握语言的基础知识。我对Java的火花有一定的了解。
我很难理解一些scala代码，因此我无法用Java编写同样的代码。我在中得到了这段代码
// Import Necessary Libraries
import com.microsoft.azure.cosmosdb.spark.schema._
import com.microsoft.azure.cosmosdb.spark._
import com.microsoft.azure.cosmosdb.spark.config.Config
// Read Configuration
val readCon
浏览 0提问于2020-09-09得票数 0
回答已采纳
我无法执行下面的代码。此代码尝试在SparkContext runJob()方法中使用SparkSession从配置单元表执行配置单元查询。
val lines = sparkSession.sparkContext.parallelize(Seq("hello world"),1)
sparkSession.sparkContext.runJob(lines, (t: TaskContext, it: Iterator[String]) => {
val conf = new SparkConf().setAppName("Testing")
val
浏览 4提问于2018-12-17得票数 1
有没有什么技术原因导致spark 2.3不支持java 1.10 (截至2018年7月)？
以下是我使用spark-submit运行SparkPi示例时的输出。
$ ./bin/spark-submit ./examples/src/main/python/pi.py
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.hadoop.security.authentication.util.KerberosUtil to me
浏览 0提问于2018-07-14得票数 20
1回答
我对Scala和Spark都很陌生，所以我希望有人能解释为什么aggregateByKey在抽象类中不能编译。这是我能想到的最简单的例子：
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
abstract class AbstractKeyCounter[K] {
def keyValPairs(): RDD[(K, String)]
def processData(): RDD[(K, Int)] = {
keyValPairs().aggregateBy
浏览 4提问于2016-10-09得票数 4
回答已采纳
1回答
我一直使用sortByKey()对元组中的键值进行排序。如果元组的大小小于9，则sortByKey函数可以正常工作。但是在大小为10的时候，sortByKey()方法失败了。如何打印大小为10或更大的元组的排序关键字。
如果我使用大小为9的密钥对RDD执行如下打印，则工作示例：
scala> freq_9s
res31: org.apache.spark.rdd.RDD[((Int, Int, Int, Int, Int, Int, Int, Int, Int), Int)]
scala> var frequent_movie_9s_count=freq_9s.count().to
浏览 5提问于2017-02-27得票数 0
回答已采纳
在写了一些MapReduce作业之后，我正在学习spark/scala。
我写了一些java beans来帮助我在HDFS中解析一个文件，我想用它来帮助我加快spark的进度。
我已经成功地加载了我的文件并创建了我的java bean对象的数组：
val input = sc.textFile("hdfs://host:port/user/test/path/out")
import my.package.Record
val clust_recs = clv_input.map(line => new my.package.Record(line))
clust_rec
浏览 1提问于2017-06-08得票数 0
1回答
使我产生了创建Scala表达式序列的需求。我似乎无法做到这一点。
我有一个SchemaRDD object z
org.apache.spark.sql.SchemaRDD =
SchemaRDD[0] at RDD at SchemaRDD.scala:103
== Query Plan ==
== Physical Plan ==
ParquetTableScan [event_type#0,timestamp#1,id#2,domain#3,code#4], (ParquetRelation ...., Some(Configuration: core-default.xml, core
浏览 0提问于2015-01-28得票数 0
我正试图将基于代码的jar打包到databricks笔记本中。
下面一行在databricks中工作，但在scala代码中抛出了一个错误：
import com.databricks.dbutils_v1.DBUtilsHolder.dbutils
val spark = SparkSession
.builder()
.appName("myApp")
.master("local")
.enableHiveSupp
浏览 7提问于2021-10-26得票数 2
1回答
我想将特定的日期格式流到Cassandra datetime列中。
我的传入日期格式为下列日期格式：
"%Y-%m-%dT%H:%M:%S.%f"
e.g. "2021-05-18T11:12:13.123456"
我的卡桑德拉桌是：
CREATE TABLE table_name (
id text,
timestamp timestamp,
PRIMARY KEY (id)
)
进食我的火花工作如下：
val df = spark.readStream
.format("kafka")
.option(
浏览 5提问于2021-05-18得票数 1
回答已采纳
1回答
我有一个关于加载火花模型的问题。这完全可以在不使用gRPC依赖的情况下工作得很好。
这是我的main.scala
import org.apache.spark.ml.tuning.CrossValidatorModel
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.Row
object lr_model {
val spark = SparkSession.builder().appName("ml model").master("local[*]")
浏览 10提问于2018-07-18得票数 0
为什么下面的结果会出现错误呢？
scala> import sqlContext.implicits._
import sqlContext.implicits._
scala> val rdd = sc.parallelize(1 to 10).map(x => (Map(x -> 0), 0))
rdd: org.apache.spark.rdd.RDD[(scala.collection.immutable.Map[Int,Int], Int)] = MapPartitionsRDD[20] at map at <console>:27
scal
浏览 0提问于2015-09-11得票数 7
回答已采纳
1回答
我试过了
./spark-2.3.1-bin-hadoop2.7/bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.1 test.py
在我自己的电脑里，一切都很好。但是当我在学校的服务器上尝试后，它有以下的信息和错误。我在谷歌搜索了很长一段时间，不知道。有谁可以帮我？
常春藤默认缓存设置为:/home/zqwang/. Ivy 2/缓存存储在以下文件中的包的jars :/home/zqwang/.Ivy 2/jars :：home设置:：url = jar:file:/data/opt/t
浏览 1提问于2018-08-27得票数 0
回答已采纳
1回答
我正在尝试下面的例子
val lista = List(("a", 3), ("a", 1), ("b", 7), ("a", 5))
val rdd = sc.parallelize(lista)
然后在shell中我得到以下信息
rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[40] at parallelize at <console>:30
但出于某种原因，我仍然没有意识到我能执行这句话
val resAgg = rdd.
浏览 4提问于2016-07-20得票数 0
我有一个关于使用scala sbt的CrossValidatorModel的问题
这是我的依赖项
libraryDependencies ++= Seq(
// spark
"org.apache.spark" %% "spark-core" % "2.3.1" % "provided" ,
"org.apache.spark" %% "spark-sql" % "2.3.1" % "provided",
"org.apache
浏览 4提问于2018-07-17得票数 1
回答已采纳
在使用Geomesa和Scala时，我一直试图使用下面的代码片段在Spark中对2列进行编码，但我不断收到一个问题，似乎Scala无法将返回的对象序列化为Dataframe。当使用Postgres和PostGIS时，生活很容易--这是一个容易的问题吗?还是有一个更好的库可以处理来自包含双格式的纬度和经度的星火数据的地理空间查询？
我在SBT中使用的版本是：
火花: 2.3.0
scala: 2.11.12
地势a: 2.2.1
jst-*：1.17.0-快照
线程“主”java.lang.UnsupportedOperationException中的异常:没有为org
浏览 1提问于2019-03-06得票数 1
我使用官方的MongoDB Spark Connector从MongoDB集合中读取Spark中的数据，代码如下：
val spark = SparkSession.
builder().
appName("MongoDB to SQL").
getOrCreate()
val df = MongoSpark.load(spark, readConfig)
df.count()
readConfig是MongoDB的标准读配置，它工作得很好。我遇到的问题是，我从MongoDB获取的一些日期/时间作为字符串
浏览 12提问于2018-11-28得票数 1
1回答
我是scala的新手，我正在尝试在我的build.sbt中导入以下库。当IntelliJ执行自动更新时，我得到以下错误：
Error while importing sbt project:
List([info] welcome to sbt 1.3.13 (Oracle Corporation Java 1.8.0_251)
[info] loading global plugins from C:\Users\diego\.sbt\1.0\plugins
[info] loading project definition from C:\Users\diego\development\
浏览 3提问于2020-06-29得票数 0
我想通过Spark从HBase加载数据，我使用hbase-spark 并引发NullPointerException
我的build.sbt文件是：
name := "proj_1"
version := "0.1"
scalaVersion := "2.11.12"
libraryDependencies ++= Seq(
"org.apache.spark" % "spark-core_2.11" % "2.3.1",
"org.apache.spark" % &
浏览 0提问于2018-09-17得票数 2
我使用Spark2.0和Scala2.11.8。
我有一个来自select查询的Cassandra ResultSet，我想将它转换为或Dataset。多么?
我一直在尝试使用这个连接器：
"com.datastax.spark" % "spark-cassandra-connector_2.11" % "2.0.0-RC1"
后来也是这样：
"com.datastax.spark" % "spark-cassandra-connector_2.11" % "2.0.0-M3"
守则：
import
浏览 5提问于2017-06-08得票数 1
回答已采纳
1回答
新的激发世界和尝试一个用scala编写的数据集示例，这是我在网上找到的。
在通过SBT运行它时，我继续获取以下错误
org.apache.spark.sql.AnalysisException: Unable to generate an encoder for inner class
你知不知道我在看什么
还可以自由地指出编写相同数据集示例的更好方法。
谢谢
> sbt> runMain DatasetExample
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.propert
浏览 5提问于2016-10-25得票数 8
回答已采纳
下面是我所有的版本
<spark.version>2.3.1</spark.version>
<scala.version>2.11.8</scala.version>
<hadoop.version>2.7.7</hadoop.version>
<artifactId>aws-java-sdk</artifactId>
<version>1.7.4</version>
我有下面的代码，它是作为脂肪罐的一部分提交的。
spark.sparkContext.hadoopCon
浏览 0提问于2018-09-13得票数 3
回答已采纳
下面有一个带有代码的笔记本，它会引发以下错误：
找不到参数sparkSession的隐式值
import org.apache.spark.sql.{SparkSession, Row, DataFrame}
import org.apache.spark.ml.clustering.KMeans
def createBalancedDataframe(df:DataFrame, reductionCount:Int)(implicit sparkSession:SparkSession) = {
val kMeans = new KMeans().setK(reducti
浏览 3提问于2019-03-15得票数 0
鉴于Groovy具有与Scala相同的大部分功能，我们是否可以编写Groovy程序来导入并在Groovy中进行科学计算。据说Apache Spark是用Scala编写的，spark为用Java或Scala编写客户端代码提供了完全相同的库。例如：
在Groovy (或Java)中
org.apache.spark.mllib.linalg.Matrix mat =
new org.apache.spark.mllib.linalg.DenseMatrix(2,2, [1,2,3,4] as double[])
在Scala中：
val mat: mat:org.apache.spark.m
浏览 1提问于2015-11-06得票数 2
最近我在Spark中使用了KernelDensity类，我试着在windows10中将它序列化到我的磁盘上，下面是我的代码：
// read sample from disk
val sample = spark.read.option("inferSchema", "true").csv("D:\\sample")
val trainX = sample.select("_c1").rdd.map(r => r.getDouble(0))
val kd = new KernelDensity().setSample(tr
浏览 1提问于2020-05-10得票数 0
这个问题针对的是熟悉py4j的人，并且可以帮助解决酸洗错误。我试图将一个方法添加到PythonMLLibAPI中，该方法接受namedtuple的RDD，做一些工作，并以RDD的形式返回结果。
该方法是以PYthonMLLibAPI.trainALSModel()方法为模型的，其类似的现有相关部分如下：
def trainALSModel(
ratingsJRDD: JavaRDD[Rating],
.. )
用于建模新代码的现有python评级类是：
class Rating(namedtuple("Rating", ["user", &#
浏览 1提问于2015-04-28得票数 14
回答已采纳
我使用的是Spark1.6.2、Hadoop2.6、Scala2.10.5和Java1.7
我使用JDBC从MSSQL读取数据，这没有任何问题：
val hqlContext = new HiveContext(sc)
val url = "jdbc:sqlserver://1.1.1.1:1111;database=CIQOwnershipProcessing;user=OwnershipUser;password=Ownership123"
val driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver
浏览 0提问于2017-01-05得票数 3
回答已采纳
在星火中使用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.1
我创建了一个DataSet，里面有MapDataType
StructType schema = new StructType(new StructField[]{
new StructField("id", DataTypes.IntegerType, false, Metadata.empty()),
new StructField("words", DataTypes.StringType, false, Metadata.empty()),
n
浏览 7提问于2017-05-22得票数 14
1回答
我有一个返回数据的UDF。就像下面的那个
scala> predict_churn(Vectors.dense(2.0,1.0,0.0,3.0,4.0,4.0,0.0,4.0,5.0,2.0))
res3: org.apache.spark.sql.DataFrame = [noprob: string, yesprob: string, pred: string]
scala> predict_churn(Vectors.dense(2.0,1.0,0.0,3.0,4.0,4.0,0.0,4.0,5.0,2.0)).show
+------------------+------
浏览 3提问于2016-12-20得票数 3
回答已采纳
对于下面的MRE，我期望得到以下输出：
$ scala mre.scala
1
object Main {
def main(args: Array[String]): Unit = {
implicit val thing = List[String]().map(_.to(List))
for (x <- 1 to 1) {
println(x)
}
}
}
然而，我得到的是以下内容
$ scala mre.scala
mre.scala:4: error: type mismatch;
found : Int(1)
requi
浏览 2提问于2021-12-11得票数 3
回答已采纳
日安，
我使用雪花文档中的示例代码来连接到它：
sfparams = {
"sfURL": "SOME_URL",
"sfUser": "SOME_USER",
"sfPassword": "SOME_PASSWORD",
"sfDatabase": "SOME_DB",
"sfSchema": "SOME_SCHEMA",
"sfW
浏览 15提问于2022-02-22得票数 0
当将星火应用程序提交给Spark时，总是会得到以下异常：
18/04/13 11:54:29 ERROR TransportResponseHandler: Still have 1 requests outstanding when connection from /10.11.9.2:6066 is closed
18/04/13 11:54:29 WARN StandaloneAppClient$ClientEndpoint: Failed to connect to master 10.11.9.2:6066
org.apache.spark.SparkException: Excep
浏览 2提问于2018-04-13得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
