Scala + Spark中字符串数组到结构数组
在Scala + Spark中，将字符串数组转换为结构数组可以通过使用Spark的内置函数和方法来实现。下面是一个完善且全面的答案：
在Scala + Spark中，将字符串数组转换为结构数组可以通过使用Spark的内置函数和方法来实现。首先，我们需要导入Spark的相关库和函数：
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._接下来，我们可以使用Spark的split函数将字符串数组拆分为多个列，并指定分隔符。例如，假设我们有一个名为data的DataFrame，其中包含一个名为strArray的字符串数组列：
val data = Seq(("1,2,3", "4,5,6"), ("7,8,9", "10,11,12")).toDF("strArray1", "strArray2")现在，我们可以使用split函数将strArray1和strArray2列拆分为多个列，并将结果存储在新的结构数组列中：
val structArray = data.select(split(col("strArray1"), ",").as("array1"), split(col("strArray2"), ",").as("array2"))这将创建一个新的DataFrame structArray，其中包含两个结构数组列array1和array2。每个结构数组列都包含原始字符串数组拆分后的多个列。
接下来，我们可以使用Spark的内置函数和方法对结构数组进行进一步的处理和操作。例如，我们可以使用getItem函数获取结构数组中的特定元素：
val element = structArray.select(col("array1").getItem(0).as("element"))这将创建一个新的DataFrame element，其中包含结构数组array1中的第一个元素。
总结一下，Scala + Spark中将字符串数组转换为结构数组的步骤如下：
- 导入Spark的相关库和函数。
- 使用
split函数将字符串数组拆分为多个列，并指定分隔符。 - 将拆分后的列存储在新的结构数组列中。
- 使用Spark的内置函数和方法对结构数组进行进一步的处理和操作。
这种转换可以在需要将字符串数组拆分为多个列进行处理和分析的情况下非常有用，例如处理CSV文件或其他包含结构化数据的文件。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云计算服务：https://cloud.tencent.com/product/cvm
- 腾讯云数据库服务：https://cloud.tencent.com/product/cdb
- 腾讯云人工智能服务：https://cloud.tencent.com/product/ai
- 腾讯云物联网服务：https://cloud.tencent.com/product/iotexplorer
- 腾讯云移动开发服务：https://cloud.tencent.com/product/mobiledv
- 腾讯云存储服务：https://cloud.tencent.com/product/cos
- 腾讯云区块链服务：https://cloud.tencent.com/product/baas
- 腾讯云元宇宙服务：https://cloud.tencent.com/product/vr
相关·内容
2回答
我想做一件简单的事。我想把所有的事件统计成2分钟的时间戳。
效果很好。
df = df.groupBy(window(df["time_value"], "2 minutes")).count()
df.show()
+--------------------+-----+
| window|count|
+--------------------+-----+
|[2018-04-10 15:00...| 770|
|[2018-04-10 00:42...| 100|
|[2018-04-10 04:14...| 54|
|[2018-04-06 15:
浏览 1提问于2018-05-15得票数 1
回答已采纳
我试图通过Scala从Elasticsearch读取数据：
Scala 2.11.8，Spark 2.3.0，Elasticsearch 5.6.8
连接-- spark2-shell --jars elasticsearch-spark-20_2.11-5.6.8.jar
val df = spark.read.format("org.elasticsearch.spark.sql").option("es.nodes", "xxxxxxx").option("es.port", "xxxx").option(&
浏览 2提问于2020-02-28得票数 2
回答已采纳
2回答
我有以下数据帧作为输出：
scala> df3.show()
+---+--------------------+
| id| app_language|
+---+--------------------+
|101|Map(tamil -> 2, t...|
|103| Map(oria -> 1)|
|107| Map(hindi -> 2)|
|102|Map(tamil -> 3, h...|
|105| Ma
浏览 0提问于2021-02-18得票数 0
1回答
我已经实现了这个代码：
scala> import org.apache.spark._
scala> import org.apache.spark.rdd.RDD
import org.apache.spark.rdd.RDD
scala> import org.apache.spark.util.IntParam
import org.apache.spark.util.IntParam
scala> import org.apache.spark.graphx._
import org.apache.spark.graphx._
scala> im
浏览 1提问于2016-08-24得票数 0
回答已采纳
1回答
得到这个问题与斯帕克2.3。
我在Cloudera集群上运行这个任务，它有7个节点: 64 GB ram，每个16核
相关conf：--conf spark.executor.memoryOverhead=5G --executor-memory 30G --num-executors 15 --executor-cores 5
火花执行器会引发错误：
java.lang.OutOfMemoryError: Requested array size exceeds VM limit
at java.lang.StringCoding.encode(StringCoding.java:350)
浏览 1提问于2018-07-18得票数 0
下面的代码试图执行一个简单的操作，尝试从稀疏向量转换为密集向量。以下是我到目前为止所拥有的：
import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature.{StringIndexer, OneHotEncoder}
import org.apache.spark.ml.feature.CountVectorizerModel
import org.apache.spark.mllib.linalg.Vector
import spark.implicits._
// Identify how many dist
浏览 3提问于2021-11-27得票数 1
回答已采纳
我想选择一个特殊元素：select("File.columns.column._name")
|-- File: array (nullable = true)
| |-- element: struct (containsNull = true)
| | |-- _Description: string (nullable = true)
| | |-- _RowTag: string (nullable = true)
| | |-- _name: string (nullable = true)
| | |
浏览 6提问于2017-07-05得票数 0
回答已采纳
1回答
给定一个数据帧，其中一列是由以下序列生成的结构序列
val df = spark
.range(10)
.map((i) => (i % 2, util.Random.nextInt(10), util.Random.nextInt(10)))
.toDF("a","b","c")
.groupBy("a")
.agg(collect_list(struct($"b",$"c")).as("my_list"))
df.printSchema
df.show(
浏览 0提问于2017-07-13得票数 12
回答已采纳
1回答
我在AWS上使用Spacy预训练的大型模型生成的文档向量时遇到了非常糟糕的时间。当我将文档向量放入一个dataframe中时，问题就开始了。
例如，这段代码对我来说很好：
# Load infrastructure libraries
import pandas as pd
import numpy as np
# Load NLP libraries and tools
import spacy
# Prepare the Spacy NLP parsers
nlp = spacy.load('en_core_web_lg')
# Load Spark
from pys
浏览 9提问于2022-09-19得票数 2
它给了我java.lang.ClassCastException：[Ljava.util.UUID；不能投给[Ljava.lang.String；
我的工作是从一个包含PostgreSQL类型列的user_ids uuid[]表中读取数据，所以当我试图保存user_ids uuid[]上的数据时，会得到上面的错误。
但是，在Cassandra上创建相同的表很好！user_ids list<text>。
我无法更改源表的类型，因为我正在读取遗留系统中的数据。
我一直在看日志上打印的点，在org.apache.spark.sql.execution.datasources.jdbc.J
浏览 1提问于2017-09-26得票数 0
当使用lib："org.apache.spark" %% "spark-sql-kafka-0-10" % "2.4.0"运行火花结构化流时，我们经常会收到当前偏移取的错误：
由: org.apache.spark.SparkException:由于阶段失败导致作业中止:阶段0.0中的任务0失败4次，最近的失败:阶段0.0中丢失的任务0.3 (TID 3，qa2-hdp-4.acuityads.org，执行器2)：断言失败:最新的offs -9223372036854775808不等于-1在org.apache.spark.sql.kafka0
浏览 3提问于2019-01-24得票数 1
回答已采纳
1回答
我试图使用Scala编写一个HDFS输出文件，并收到以下错误：
线程"main“中的异常:在org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:315) at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:305) at org.apache.spark.util.ClosureCleaner$.clean(ClosureCl
浏览 4提问于2016-05-13得票数 0
我正在尝试使用变量将记录插入到表中，但失败了。 命令： val query = "INSERT into TABLE Feed_metadata_s2 values ('LOGS','RUN_DATE',{} )".format(s"$RUN_DATE")
spark.sql(s"query")
spark.sql("INSERT into TABLE Feed_metadata_s2 values ('LOGS','ExtractStartTimestamp',$Extr
浏览 36提问于2021-04-26得票数 0
回答已采纳
我的主要目标是连接Kafka，创建一个DStream，将其作为行保存到局部变量并将其写入mongo，并在PySpark中具有端到端流。
但是，在创建DStream时，我在第一步就面临问题，错误是"java.util.ArrayList不能转换为java.lang.String“。你能帮我确认一下修复程序吗？详情如下，
我试着用下面的电火花连接卡夫卡，
kafkaParams = {"metadata.broker.list": ['host1:port','host2:port','host3:port'],
"s
浏览 1提问于2020-02-12得票数 1
回答已采纳
--我尝试从Socket获取数据以附加到dataframe --我接收了数据并将其执行到Seq()中，但是当我使用forEach将它们附加到dataframe时，有一个问题--这是我的代码：
object CustomReceiver {
def main(args: Array[String]): Unit = {
StreamingExamples.setStreamingLogLevels()
// Create the context with a 1 second batch size
val spark: SparkSession = SparkS
浏览 2提问于2021-06-21得票数 0
回答已采纳
这个GC开销限制错误让我发疯了。我有20个执行者使用25 GB，我完全不明白它怎么能抛出GC开销，我也不是那个大数据集。一旦这个GC错误发生在执行器中，它就会丢失，其他执行程序也会慢慢丢失，因为IOException、Rpc客户端断开、洗牌找不到等等。
我是星火的新手。
WARN scheduler.TaskSetManager: Lost task 7.0 in stage 363.0 (TID 3373, myhost.com): java.lang.OutOfMemoryError: GC overhead limit exceeded
at org.apache
浏览 0提问于2015-08-18得票数 1
回答已采纳
互联网上有几个类似的问题，但没有人知道答案。
我使用下面的代码将mongo数据保存到Hive，但是异常会发生，如最后所示。我会问如何解决这个问题。
我在用
火花-蒙戈连接器(火花2.1.0 - scala 2.11)
java-mongo驱动程序3.10.2
导入com.mongodb.spark.MongoSpark org.apache.spark.SparkConf org.apache.spark.sql.SparkSession org.apache.spark.sql.SparkSession org.apache.spark.sql.types.StructType Mo
浏览 6提问于2019-04-27得票数 0
2回答
我需要在火花读取文本文件到datasetT。该文件的格式不正确，因为它有一些空白字段，并且很难定义拆分字符串的参数。我一直在尝试将数据读取到RDD中，然后将其转换为case类类型，然而，并不是所有的字段都被正确解析，我得到了一个错误：
java.lang.NumberFormatException: empty String
at sun.misc.FloatingDecimal.readJavaFormatString(FloatingDecimal.java:1842)
at sun.misc.FloatingDecimal.parseDouble(Floa
浏览 11提问于2019-06-08得票数 3
我正在尝试构建一个简单的基于Scala的Spark应用程序并在电子病历中运行它，但是当我运行它时，我得到的是Error: Failed to load class: com.myorganization.MyScalaObj。我的Scala文件是： package com.myorganization
import org.apache.spark.sql.SparkSession
object MyScalaObj extends App {
val spark = SparkSession.builder()
.master(("local[*]"))
浏览 17提问于2020-10-02得票数 1
回答已采纳
我有一个名为outLinks的数据集，它有两列，一个字符串列和一个数组列。如下所示：
+-------+---------------------+
| url|collect_set(outlinks)|
+-------+---------------------+
| link91| [link620, link761]|
|link297| [link999, link942...|
|link246| [link623, link605...|
...
我正在尝试向该表追加多个行，其中每个新行都包含一个字符串和一个空列表。diff是一个包含一个字符串列的数据集。
outLi
浏览 10提问于2017-12-09得票数 0
回答已采纳
我正试着从Spark数据帧中写给Cassandra。当我有一个简单的数据帧模式时，如示例中所示，它可以工作：
root
|-- id: string (nullable = true)
|-- url: string (nullable = true)
但是，当我尝试编写包含StructTypes的数据帧时，其模式如下：
root
|-- crawl: struct (nullable = true)
| |-- id: string (nullable = true)
然后我得到以下异常：
Exception in thread "main" java.lang
浏览 4提问于2016-06-27得票数 1
2回答
使用PySpark 1.6.3，我试图将RDD转换为Dataframe。这是在齐柏林飞艇笔记本上运行的测试代码。感兴趣的是rdd_ret。
>>> from pyspark.sql import Row
>>> rdd_ret.count()
9301
>>> rddofrows = rdd_ret.map(lambda x: Row(**x))
>>> things = rddofrows.take(10000)
>>> len(things)
9301
>>> [type(x) fo
浏览 0提问于2018-06-21得票数 2
我正在做以下工作：
使用新4j火花连接器从neo4j中获取数据
使用apache火花连接器从hbase中获取数据
通过df.show()将两种数据打印到控制台
使用spark连接两个数据帧，使用两个数据文件中存在的唯一id列：
Dataset mergedData = ss.sql("SELECT * from hbasetable，neo4jtable WHERE hbasetable.nodeId = neo4jtable.id")；
通过mergedData.show()将数据打印连接到控制台
一切都很顺利。但是，现在，我更改了用于获取neo4
浏览 1提问于2018-06-04得票数 1
我有一个情绪分析程序，可以使用递归神经网络来预测给定的电影评论是正面的还是负面的。我正在使用Deeplearning4j深度学习库来实现这个程序。现在，我需要将该程序添加到apache spark管道中。
这样做的时候，我有一个扩展org.apache.spark.ml.classification.ProbabilisticClassifier的类MovieReviewClassifier，我必须在管道中添加该类的一个实例。使用setFeaturesCol(String s)方法将建立模型所需的特征输入程序。我添加的功能是String格式的，因为它们是一组用于情感分析的字符串。但是这些特性应
浏览 3提问于2016-02-19得票数 1
我试图从Elasticsearch中读取数据，并将数据转储到Kafka主题中。当我写卡夫卡主题的时候，我得到了ClassCastException
我正在使用Spark2.3.2和Scala2.11.2。我正在运行下面的代码，并在ClassCastException中结束。我不知道怎么解决这个问题。同样的代码(从斯帕克写到卡夫卡)似乎也适用于其他人()。
import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import spark.implicits._
val reader = spark.read
浏览 20提问于2022-05-02得票数 0
我尝试创建一个简单的火花会话，它读取一个csv文件。但它告诉我这个错误：
Exception in thread "main" java.lang.reflect.InaccessibleObjectException: Unable to make field private transient java.lang.String java.net.URI.scheme accessible: module java.base does not "opens java.net" to unnamed module @bcec361
at java.base/ja
浏览 48提问于2022-07-06得票数 0
我有一份数据。我的Hive查询结果总是只有一条记录，只有一列。有时它产生null，有时它产生0，有时它产生一个数字(BigInt)
我的问题是，我想将这个单一值的结果存储到scala中的val中。我该怎么做？
scala> val hc = new HiveContext(sc)
hc: org.apache.spark.sql.hive.HiveContext = org.apache.spark.sql.hive.HiveContext@19e15d63
scala> val df = hc.sql("""select MAX(seq_num) as
浏览 0提问于2020-04-02得票数 0
回答已采纳
我找了又找，但我找不到答案。
我有一个函数，它本质上将列(根据它们的名称和数据类型)排序为我的公司在将DataFrames保存到SQL之前所需的特定顺序。我知道这个功能起作用。它所做的就是输出一个列列表(在源DataFrame中找到)。我调用这个排序列名列表col_list，并使用它作为df.select()的参数。
这种方法几个月来一直行之有效。但是，在一个特定的项目中，当我试图通过以下方法将排序应用于数据访问时：
df = df.select(*col_list)
...I得到了以下错误：
org.apache.spark.SparkException: Job aborted due t
浏览 11提问于2022-04-13得票数 0
我正在尝试使用Apache-Spark来并行处理目录中分离的csv文件。具体地说，我希望每个从节点将每个csv文件中第一列的所有编号相加，并返回计算结果。以下是我的代码：
import os, sys, inspect, csv
### Current directory path.
curr_dir = os.path.split(inspect.getfile(inspect.currentframe()))[0]
### Setup the environment variables
spark_home_dir = os.path.realpath(os.path.abspath(
浏览 1提问于2015-07-15得票数 1
我有下面这样的数据。在一个数组中，我们有不同的单词
scala> val x=rdd.flatMap(_.split(" "))
x: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[9] at flatMap at <console>:26
scala> x.collect
res41: Array[String] = Array(Roses, are, red, Violets, are, blue)
我想在scala中找到数组中每个单词的长度
浏览 2提问于2019-11-26得票数 0
1回答
我希望在用户设计的函数中使用位于另一个类中的方法，但它不起作用。
我有一个方法：
def traitementDataFrameEleve(sc:SparkSession, dfRedis:DataFrame, domainMail:String, dir:String):Boolean ={
def loginUDF = udf((sn: String, givenName:String) => {
LoginClass.GenerateloginPersone(sn,givenName,dfr)
})
dfEleve.
浏览 0提问于2017-04-25得票数 0
帮助请修复此错误：
20/04/09 18:38:44 ERROR MicroBatchExecution: Query [id = 9f3cbbf6-85a8-4aed-89c6-f5d3ff9c40fa, runId = 73c071c6-e222-4760-a750-393666a298af] terminated with error
java.lang.ClassCastException: org.apache.spark.sql.execution.streaming.SerializedOffset cannot be cast to org.apache.spark.sql.
浏览 10提问于2020-04-09得票数 0
1回答
从一个查询中的多个主键中提取条目
、、、、
我正在使用火花卡桑德拉连接器(最新版本，我认为2.0.3)从Cassandra (3.10)提取数据到Spark2。我想知道是否有一种方法从一个查询中提取多个主键。PKs是原子的(只有一个字段定义了键)。但是它们是散列，所以我不能进行范围查询。有没有办法提取
val set = Set(hash1,hash2,hash3)
在一个呼叫sc.cassandraTable("keyspace","table").where("id =?",set)或其他类似的东西。
我把Scala和火花结合使用。
编辑：
一项建议是采用以下方法
sc.cassand
浏览 2提问于2017-08-10得票数 2
回答已采纳
1回答
我正在多次运行一个涉及spark并行化的程序。这个程序在最初的几次迭代中运行良好，但由于内存问题而崩溃。我使用的是Spark 2.2.0和Python2.7，我在亚马逊EC2上运行我的测试，内存为30g。
下面是我的Spark设置：
conf = pyspark.SparkConf()
conf.set("spark.executor.memory", '4g')
conf.set('spark.executor.cores', '16')
conf.set('spark.cores.max', '16&#
浏览 0提问于2017-10-12得票数 1
我试图从卡夫卡读取数据使用火花结构化流和预测形式的传入数据。我用的是我用Spark训练的模型。
val spark = SparkSession
.builder()
.appName("Spark SQL basic example")
.master("local")
.getOrCreate()
import spark.implicits._
val toString = udf((payload: Array[Byte]) => new String(payload))
val sentenceDataFrame = spark
浏览 7提问于2017-10-03得票数 2
回答已采纳
我试图使用结构化流从Apache中的MQTT代理读取json流，读取传入json的一些属性并将它们输出到控制台。我的代码是这样的：
val spark = SparkSession
.builder()
.appName("BahirStructuredStreaming")
.master("local[*]")
.getOrCreate()
import spark.implicits._
val topic = "temp"
val brokerUrl = "tcp://localhost:1883"
v
浏览 0提问于2020-11-13得票数 1
回答已采纳
这个问题已经被问到了Scala的，但它对我没有帮助，因为我正在使用Java API。我真的把所有的东西和厨房的水槽都扔进去了，所以这是我的方法：
List<String> sourceClasses = new ArrayList<String>();
//Add elements
List<String> targetClasses = new ArrayList<String>();
//Add elements
dataset = dataset.withColumn("Transformer", callUDF(
浏览 40提问于2019-11-25得票数 1
回答已采纳
在为CCO调用Apache Mahout的SimilarityAnalysis时，我得到了一个关于NegativeArraySizeException的致命异常。
我运行的代码如下所示：
val result = SimilarityAnalysis.cooccurrencesIDSs(myIndexedDataSet:Array[IndexedDataset],
randomSeed = 1234,
maxInterestingItemsPerThing = 3,
maxNumInteractions = 4)
我看到了以下错误和相应的堆栈跟踪：
17/0
浏览 28提问于2017-04-20得票数 1
我对Spark和Scala并不熟悉--我被困在这个异常中，我试图添加一些额外的字段，即StructField到从数据框架中检索到的现有StructType中，该列使用Spark并在异常下面进行gettting。
代码片段：
val dfStruct:StructType=parquetDf.select("columnname").schema
dfStruct.add("newField","IntegerType",true)
线程"main“中出现异常
org.apache.spark.sql.types.DataTypeExce
浏览 2提问于2016-03-30得票数 1
回答已采纳
我试着遵循这里的教程：。使用命令sbt package，我获得了一个空jar。随后的命令：
spark-submit --class SimpleApp --master local[4] target/scala-2.11/simple-project_2.11-1.0.jar
在以下方面的成果：
java.lang.ClassNotFoundException: SimpleApp
at java.net.URLClassLoader.findClass(Unknown Source)
at java.lang.ClassLoader.loadClass(U
浏览 2提问于2016-10-31得票数 1
回答已采纳
我想运行火花-核心run ，但我得到一个java.lang.ClassNotFoundException错误时运行火花提交。
下面是github示例中的scala代码，我将其放入一个对象中，并定义了一个SparkContext。
analyzer.Sentiment.scala：
package analyzer
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.sql.func
浏览 2提问于2016-06-22得票数 1
回答已采纳
1回答
我已经从3.1.1版升级到3.2.1版。现在，所有现有的火花作业都会出现以下错误。
Exception in thread "main" org.apache.spark.SparkException: Driver cores must be a positive number
at org.apache.spark.deploy.SparkSubmitArguments.error(SparkSubmitArguments.scala:634)
at org.apache.spark.deploy.SparkSubmitArguments.validateSubmitArg
浏览 4提问于2022-03-31得票数 2
我正在尝试从多个路径读取一些avro文件到DataFrame。假设我的路径是这个路径下的"s3a://bucket_name/path/to/file/year=18/month=11/day=01"，我还有两个分区，假设是country=XX/region=XX
我想一次读取多个日期，而不需要明确命名国家和地区分区。此外，我希望国家和地区是这个DataFrame中的列。
sqlContext.read.format("com.databricks.spark.avro").load("s3a://bucket_name/path/to/file/yea
浏览 2提问于2018-12-03得票数 2
我正在尝试运行一个简单的程序，在scala中用火花来计数单词。我已经在linux中自己完成了所有的安装，并且我无法执行它，因为我有以下错误：
java.lang.ClassNotFoundException: scala.Function0
at sbt.internal.inc.classpath.ClasspathFilter.loadClass(ClassLoaders.scala:74)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at java.lang.Class.forName0(Native Method)
浏览 3提问于2017-10-26得票数 0
回答已采纳
几个小时以来，我一直试图让这段代码正常工作：
val spark = SparkSession.builder()
.appName("Consumer")
.getOrCreate()
spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", url)
.option("subscribe", topic)
.load()
.select("value")
.writeStream
.format(fileFor
浏览 0提问于2018-11-20得票数 0
回答已采纳
下面是代码
val path = "C:\\Users\\John\\Downloads\\crimes.csv"
val crimeFile = sc.textFile(path)
val crimerows = crimeFile.map(l=>l.split(",").map(e=>e.trim))
//taking the first row as header
val header = crimerows.first
//filter out the header
val crimes = crimerows.filter(_(0)!=he
浏览 2提问于2015-04-28得票数 2
回答已采纳
我正在与<spark.version>2.2.1</spark.version>合作，我想写一个数据帧，其中有一个映射字段到postgres作为json字段。
示例代码：
import java.util.Properties
import org.apache.spark.SparkConf
import org.apache.spark.sql.{SaveMode, SparkSession}
import scala.collection.immutable.HashMap
case class ExampleJson(map: HashMap[String
浏览 44提问于2021-05-19得票数 1
回答已采纳
我正在使用PostGre作为数据库。我希望为每批获取一个表数据，并将其转换为拼花文件并存储到s3中。我试图使用spark和readStream的JDBC连接，如下所示.
val jdbcDF = spark.readStream
.format("jdbc")
.option("url", "jdbc:postgresql://myserver:5432/mydatabase")
.option("dbtable", "database.schema.table")
.option(
浏览 1提问于2019-07-02得票数 5
我想使用星火流，并连接它与卡夫卡。然而，我仍然得到了NoSuchMethodError: NoSuchMethodError错误，现在我只是不知道下一步该做什么。
我的设置：
Ubuntu 16.04
Scala 2.11
Kafka 2.11-1.0.0 (我也尝试使用2.11-0.10.0.0)
火花2.2.1
Hadoop 2.9.0
我甚至不能运行示例脚本：
from pyspark.sql import SparkSession, Row, SQLContext
from pyspark import SparkContext, SparkConf
from pyspark.stre
浏览 1提问于2018-02-09得票数 3
我正在尝试获取卡夫卡流数据，这是火花流中的JSONArray，每个JSONArray包含几个JSONObject。
我希望将每个JSONObject保存到datadrames中，并在与另一个表映射后保存到cassandra表中。
我试图创建dataframe来保存JSONObject，但是当我在stream.foreachRDD中创建dataframe时，它抛出了NullPointerException。是因为spark不支持嵌套的RDD吗？如果是这样，我如何将JSONObject保存到cassandra？
数据格式如下：
[
{
"temperature&#
浏览 1提问于2016-12-12得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
