从 SQL Server 到 MongoDB Atlas 的 ETL 管道答案

【问题标题】：ETL pipeline from SQL Server to MongoDB Atlas从 SQL Server 到 MongoDB Atlas 的 ETL 管道
【发布时间】：2021-07-23 06:07:14
【问题描述】：

我需要您的 cmets 了解如何创建从 SQL Server 到 MongoDB Atlas 云的数据管道。在这个管道中，我做了许多复杂的转换甚至分析和故障检测任务，需要将当前数据与之前处理的数据进行比较。有一些错误标准只能在将新数据与之前处理的数据进行比较时计算出来。因此，管道不仅仅是单一方向的。是否可以使用 AWS 上的 Kinesis-Lambda 管道完成，或者最好使用我们公司服务器上的 Kafka 和 Spark 完成，然后将结果上传到云端？在这两种情况下，我的转换单元如何从目标读取数据。

我在下图中描述了我的两个想法。

【问题讨论】：

Kafka 和 Kinesis 都可以处理高吞吐量数据流。但是你需要这个吗，因为你的是 ETL 场景？
您没有提到任何性能要求。面对这些类型的架构问题，我总是问自己“如果我离开公司，新员工更容易理解和支持什么”。
@RajdeepDebnath Kafka 和 Kinesis 将只进行数据流和解耦。我的主要转换功能发生在 Lambda 或 Spark 中。我想知道哪个曲目更有意义。
@barrypicker 感谢您的评论。对我来说，下一个人的学习曲线目前不是问题。我担心的是上层路径是否可能？例如 Lambda 应该由 Kinesis 中的新数据之类的东西触发；那么我可以同时从源 (Kinesis) 和目标 (MongoDB) 读取数据来进行比较和分析吗？

标签： sql-server mongodb apache-spark aws-lambda pipeline

【解决方案1】：

我会为所有人使用 spark。从 sqlserver 读取，进行转换并写入 mongo (sql server --> spark --> mongoDB)：

#SQL server    
df = (
        spark
            .read
            .format('jdbc')
            .option("driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver")
            .option('url', 'jdbc url')
            .option('user', user)
            .option('password', password)
            .option('dbtable', 'schema.table')
            .option('tempdir', 's3://....')
            .option('forward_spark_s3_credentials', 'true')
            .load()
    
        )
    
    # Mongo
df = spark.read.format("mongo").option("uri","mongodb://127.0.0.1/people.contacts").load()
df.write.format("mongo").option("uri","mongodb://127.0.0.1/people.contacts").mode("append").save()

【讨论】：