如何使用spark/scala将json字符串格式化为MongoDB文档样式？
使用Spark/Scala将JSON字符串格式化为MongoDB文档样式的方法如下：
- 导入相关依赖：
import org.apache.spark.sql.{SparkSession, SaveMode}
import org.apache.spark.sql.functions._- 创建SparkSession：
val spark = SparkSession.builder()
.appName("JSON to MongoDB")
.config("spark.mongodb.output.uri", "mongodb://localhost/mydb.collection")
.getOrCreate()其中，mongodb://localhost/mydb.collection是MongoDB的连接URI，mydb.collection是要写入的数据库和集合名称。
- 定义JSON字符串：
val jsonString = """{"name": "John", "age": 30, "city": "New York"}"""- 将JSON字符串转换为DataFrame：
val jsonDF = spark.read.json(Seq(jsonString).toDS())- 格式化DataFrame为MongoDB文档样式：
val formattedDF = jsonDF.withColumn("_id", monotonically_increasing_id())这里使用monotonically_increasing_id()函数为每个文档生成唯一的_id字段。
- 将DataFrame写入MongoDB：
formattedDF.write
.format("com.mongodb.spark.sql.DefaultSource")
.mode(SaveMode.Append)
.save()完整的代码示例：
import org.apache.spark.sql.{SparkSession, SaveMode}
import org.apache.spark.sql.functions._
val spark = SparkSession.builder()
.appName("JSON to MongoDB")
.config("spark.mongodb.output.uri", "mongodb://localhost/mydb.collection")
.getOrCreate()
val jsonString = """{"name": "John", "age": 30, "city": "New York"}"""
val jsonDF = spark.read.json(Seq(jsonString).toDS())
val formattedDF = jsonDF.withColumn("_id", monotonically_increasing_id())
formattedDF.write
.format("com.mongodb.spark.sql.DefaultSource")
.mode(SaveMode.Append)
.save()注意：在运行代码之前，需要确保已经正确配置了Spark和MongoDB的环境，并且已经添加了相关的依赖。
推荐的腾讯云相关产品：腾讯云数据库 MongoDB，产品介绍链接地址：https://cloud.tencent.com/product/mongodb
相关·内容
我有一个包含两列id、json_string的表，需要将json_string转换为MongoDB文档格式。我正在从Spark/Scala向MongoDB发送数据。我试着使用withColumn，但是我仍然不能得到想要的格式。这就是我到目前为止所拥有的，所以任何帮助都会非常感谢。原始json字符串示例(df) val df=spark.sql("
浏览 31提问于2021-04-15得票数 0
回答已采纳
我目前有一个由两列组成的csv："abc","123""def","123"{"abc":["123","234"],"def":["123"]}
我创建了一个数据框架
浏览 5提问于2022-09-13得票数 1
我正在尝试从MongoDB读取一个庞大的复杂文档到spark数据框架中。当我将这个数据库转换成json时，它就能工作了。但是，如果我直接从com.mongodb.spark.exceptions.MongoTypeConversionException:中读取，则会得到以下错误:由: MongoDB引起的不能将字符串转换为当我试图显示或写入json/csv时会出错。mongo$.main(mongo.<em
浏览 1提问于2018-04-03得票数 1
我使用的是Mongo连接器--文档中的所有示例()都很好，但是如果我在一个有子文档失败的文档中测试查询，显然SQL还没有准备好进行此查询： at com.mongodb.spark.
浏览 5提问于2016-08-16得票数 0
回答已采纳
我使用下面的代码将mongo数据保存到Hive，但是异常会发生，如最后所示。我会问如何解决这个问题。我在用
导入com.mongodb.spark.MongoSpark org.apache.spark.SparkConf$.apply(BsonValueToJson.scala:29)
at com.mongodb
浏览 6提问于2019-04-27得票数 0
当我试图读取不包含的JSON字符串时，我会收到一个字段丢失的异常。spark.read.json(Seq("{'dummy': '1', 'b': {'a': '1'}}").toDS).as[X] at org.apache
浏览 0提问于2020-02-04得票数 3
回答已采纳
使用最新的MongoDB连接器用于Spark (v10)并尝试连接两个数据格式将产生以下无益错误。.DataSourceV2Relation.name(DataSourceV2Relation.scala:66)Pyspark代码只是简单
浏览 18提问于2022-05-11得票数 3
1回答
假设我有一个wordcount示例，其中我在一列中以word的形式获得数据，而在另一列中获得wordcount，则希望收集相同的数据，并将其存储为mongo集合中的json数组。abc",count:1},{word:"xyz",count:23}]}explodedWords1.toJSON.toDF("wo
浏览 2提问于2018-04-13得票数 3
我正在尝试使用SQLContext.read()读取由Kafka在Spark中生成的JSON记录。每次NullPointerException出现。<init>(DataFrameReader.scala:535) at org.apache.spark</
浏览 1提问于2016-08-10得票数 3
1回答
我想把它简化为一个值，然后在我的程序中使用这个值。“).load().selectExpr(“将值转换为字符串)为json").reduce { (x，y) =>}
我期望将流简
浏览 0提问于2019-07-20得票数 0
回答已采纳
2回答
我使用bson Document类来创建json字符串(对于mongodb)。
但是json字符串并不花哨。一切都在一条线上。如何将json字符串格式化为花哨的(多行) json字符串？
浏览 2提问于2016-03-05得票数 1
2回答
我正在使用Ruby在MongoDB中打印一个BSON文档。这是使用用于MongoDB的Ruby包装器完成的。client[:test].find().each do |doc| end
find方法返回一个BSON文档。这可以是到一个字符串使用to_json方法。虽然我可以使用regex魔术将上面的字符串格式<em
浏览 3提问于2018-01-21得票数 0
回答已采纳
我是spark的新手。我使用结构化流媒体从kafka中读取数据。我可以在Scala中使用以下代码读取数据： .format("kafka")
.option("kafka.bootstrap.servers流式api以二进制格式提供数据。我看到了将数据转换为string或json的示例，但我找不到任何如何将数
浏览 1提问于2017-10-19得票数 2
我使用的是play framework 2.2.0，并使用salat为play+ scala配置了mongo。通过在build.sbt中添加以下内容，我将spark与上面的内容集成在一起"com.typesafe.akka" % "1.3.0",
"org.apach
浏览 1提问于2014-12-10得票数 2
我试图向MongoDB写入一个Spark，它包含一个ObjectId的字符串表示形式，它是另一个集合的_id。关键是，PySpark不支持ObjectId (Scala和Java ObjectId支持在这里解释：)，所以，如何使用Spark从PySpark将ObjectId插入MongoDB？
浏览 3提问于2017-11-23得票数 3
回答已采纳
我试图使用MongoDB将一些JSON从Spark (Scala)保存到MongoDB。我遇到的问题是，这个API似乎总是将数据保存为"{_id：.，value：{}“。在下面的代码示例中，我的文档保存如下： "_id" : ObjectId("55e80cfea9fbee30aa703261"), &qu
浏览 4提问于2015-09-03得票数 3
我使用Apache将少量(<10)的JSON文档写入Cosmos DB作为概念的证明 at com.mongodb.spark.MongoSpark$$anonfun$save$1$$anonfun$apply$1.apply(MongoSpark.scalacom.mongodb.spark.MongoConnect
浏览 5提问于2019-11-15得票数 0
1回答
我有一个JSON文件，我希望使用Spark将其加载到MongoDB中。我采用了将单个元素加载到集合中的方法，如下所示val collection = mongoClient(description" -> s"description $a", "name&qu
浏览 7提问于2016-02-22得票数 0
当配置单元表有像映射或数组这样的值时，如果你在配置单元客户端选择它，它们会显示为JSON，例如：{"a":1,"b":1}或[1,2,2]。在Spark中选择这些对象时，它们是DataFrame中的贴图/数组对象。如果对每一行进行字符串化，则它们是Map("a" -> 1, "b" -> 1)或WrappedArray(1, 2, 2)。当我使用Spark的HiveContext时，我想使用</e
浏览 1提问于2018-07-20得票数 0
当我尝试将一系列文档写入MongoDB时，我得到了这样一个异常：java.lang.IllegalArgumentException: clusterListener can not be nullval sparkVersion = "org.twitter4j&
浏览 2提问于2018-05-27得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
