mongodb - 将数据从无模式数据库迁移到关系数据库 : MongoDB to Snowflake

Question

我们有一个用例促使我写这篇文章，我相信你们中的许多人都会遇到这种情况。这种情况是通过单个 Talend 作业将多个集合从 MongoDB 迁移到 Snowflake 数据库，并将集合的顶级节点保留为 Snowflake 表中的单个字段。

现在我们知道 Talend 不支持 MongoDB 源的动态模式，因为 MongoDB 集合不强制模式，这意味着我们必须为我们想要的每个现有/新集合创建单独的作业/子作业我们还必须重新设计作业以应对文档中的 future 更改，同时确保它始终有效，因此我们必须寻找替代解决方案。

方法是这样的，

第一步:从 MongoDB 集合中获取所有顶级键及其类型。我们使用 $objectToArrray 聚合将所有顶级键和值对转换为文档数组，然后是 $unwind 和 $group 以及 $addToSet 在整个集合中获取不同的键和值类型。

 {
"_id" : "1",
"keys" : [ 
    "field1~string", 
    "field2~object", 
    "filed3~date",
    "_id~objectId"
 ]

第二步:在 Mongo 数据类型和 Snowflake 数据类型之间创建一对一映射。我们创建了一个名为“dataTypes”的 HashMap 来存储此信息。或者，此信息可以存储在表格或文件等中。

 java.util.Map<String,String> dataTypes = new java.util.HashMap<String,String>();
 dataTypes.put("string","VARCHAR");
 dataTypes.put("int","NUMBER");
 dataTypes.put("objectId","VARCHAR");
 dataTypes.put("object","VARIANT");
 dataTypes.put("date","TIMESTAMP_LTZ");
 dataTypes.put("array","VARCHAR");
 dataTypes.put("bool","BOOLEAN");

第三步:将键与雪花进行比较:首先我们查询雪花 INFORMATION_SCHEMA 表是否存在，如果不存在则创建表，如果存在则检查更改文档中的字段并添加或修改雪花表中的那些列。 DDL 脚本是通过在第二步中使用“数据类型映射”并在第一步中迭代键来生成的

第四步:使用 mongoexport 命令将数据从 MongoDB 卸载到本地文件系统:

mongoexport --db <databaseName> --collection <collectionName> --type=csv --fields=<fieldList> --out <filename>

这是根据第一步中的键准备的。

第五步:使用 Snowsql 使用 PUT 命令将 .csv 文件从本地文件系统暂存到雪花暂存位置。

snowsql -d <database> -s <schema> -o exit_on_error=true -o log_level=DEBUG  -q  'put <fileName> @<internalStage> OVERWRITE=TRUE';

第六步:将暂存位置的数据加载到雪花表

COPY INTO <tableName> FROM @<internalStage> 
[file_format=<fileFormat>] [pattern=<regex_pattern>]

在这里指定 file_format 和模式是可选的，我们使用了正则表达式，因为我们在一个雪花阶段为每个集合暂存多个文件。

第七步:维护集合列表，该列表可以放在本地文件系统或数据库表中的文件中，在 Talend 作业中迭代集合列表并通过参数化集合名称通过上述步骤处理每个集合、表名、文件名和暂存名等。

Answer 1

一种解决方案是将您的 Mongodb 集合的记录加载到 variant 的 Snowflake 字段中类型。然后，创建一个雪花 view使用 Snowflake 的 dot notation 提取特定 key .

将您的数据导出为 JSON 类型。

mongoexport --type=json --out <filename>

将该导出加载到具有如下结构的表中。

create table collection_name_exports (
  data variant,  -- This column will contain your export
  inserted_at datetime default current_timestamp()
);

根据需要将键提取到 View 的列中。

create view collection_name_view as
select
  collection_name_exports:key1 as field1,
  collection_name_exports:key2 as field2
from collection_name_exports