问使用PySpark实现MongoDB到Elasticsearch

EN

在Spark中，你可以使用RDDs或DataFrames。RDDs不是结构化的，是较旧的Spark API。要使用RDDs，您需要使用MapPartition、ForEachPartition等。DataFrames是使用类似于ANSI SQL的Spark SQL函数来构造和启用的。

MongoDB支持每个文档的模式。Spark DataFrames要求每个DataFrame都有一个统一的模式。

如果您的集合对每个集合的所有文档都有一个统一的模式(取决于插入的数据是什么)，您可以手动(通过代码)创建Spark模式定义(PySpark中的pyspark.sql.types.StructType)。例如，查看StructType、DataType、StructDef、FieldType、ArrayDef、DictDef、StringType等的Spark文档。

如果您的Elasticsearch集合没有统一的模式，则需要使用RDDs，然后通过MapPartition将其转换为统一的模式，以便将此统一模式用于MongoDB索引。

您可以使用Spark SQL通过pyspark使用：spark_session.read.format('mongo').option('uri', 'mongodb://uri.to.mongo.goes.here').schema(schema=spark_schema_goes_here)从Mongo读取DataFrames，并使用df.write.format('mongo').option('uri', uri).mode(write_mode_eg_append).save()进行类似的写入

我不记得Elastic语法了，但是，它是相似的，并且定义是每个索引的模式。尝试在MongoDB站点和Elasticsearch站点中搜索Spark以查找更多详细信息。

可以从Spark模式定义自动创建Elastic索引定义，我曾经这样做过，但是没有代码。