数据仓库数据填充与操作指南

1. 数据加载验证与更新测试

1.1 数据加载验证

在进行数据加载操作后，我们需要验证所有数据是否已正确加载到 NDS 中。可以通过查询阶段表（stage）和 NDS 状态表来完成此验证。以下查询用于测试阶段表中的每一行是否存在于 NDS 中：

select * from stage.dbo.state a
where not exists
( select * from nds.dbo.state b
where a.state_code = b.state_code )

如果此查询没有返回任何行，则表示阶段表中的所有行都已成功加载到 NDS 中。

1.2 数据更新测试

在完成“初始加载”后，我们需要测试阶段数据的更改是否能通过之前的 SCD 转换正确加载到 NDS 中。以更新亚利桑那州（Arizona）的人口数据为例，操作步骤如下：
1. 在 SQL Server Management Studio 中，依次执行以下 SQL 语句更新阶段数据库中的数据：

use stage
select population from state where state_code = 'AZ' --6166318
update state set population = 6167318 where state_code = 'AZ'

2. 在运行 SSIS 包之前，查询 NDS 中亚利桑那州的人口数据：

select population from nds.dbo.state where state_code = 'AZ'

此时，NDS 中的人口数据应为 6,166,318。
3. 运行 SSIS 包，然后再次运行上述 NDS 查询。此时，NDS 中的人口数据应更新为 6,167,318。
4. 将阶段数据库中的人口数据恢复为 6,166,318，再次运行 SSIS 包，使数据恢复原状。

2. 使用 SQL 和查找操作进行 Upsert

2.1 Upsert 概述

之前的 SCD 转换实际上就是一种常见的 Upsert 操作，即“存在则更新，不存在则插入”。这是数据仓库中的基本操作，在构建数据仓库 ETL 系统时会经常遇到。在 SSIS 中，常用的 Upsert 方法有两种：使用 SQL 语句和使用查找转换。

2.2 使用 SQL 语句进行 Upsert

以下是使用 SQL 语句进行 Upsert 操作的示例，以国家表为例：

-- 更新 NDS 表中与阶段表中国家名称不同的行
update nds.dbo.country
set country_name = s.country_name,
source_system_code = 2,
update_timestamp = getdate()
from the stage.dbo.country s
inner join nds.dbo.country n
on n.country_code = s.country_code
where n.country_name <> s.country_name
and n.country_key <> 0

-- 插入阶段表中不存在于 NDS 的行
insert into nds.dbo.country
( country_code, country_name, source_system_code,
create_timestamp, update_timestamp )
select s.country_code, s.country_name,
2 as source_system_code,
getdate() as create_timestamp,
getdate() as update_timestamp
from the stage.dbo.country s
left join nds.dbo.country n
on s.country_code = n.country_code
where n.country_key is null

在 SSIS 中实现上述 SQL 语句的操作步骤如下：
1. 创建一个新的 SSIS 包，并将其重命名为“NDS country using SQL”。
2. 从工具箱中拖动“Execute SQL 任务”到设计界面，将其重命名为“Update NDS Country”。
3. 再拖动一个“Execute SQL 任务”到设计界面，将其重命名为“Insert NDS Country”。
4. 将两个任务的连接设置为“servername.NDS.ETL”。
5. 将 SQL 语句设置为上述语句，并将“SQLSourceType”保留为“直接输入”。
6. 将“Update NDS Country”框的绿色箭头连接到“Insert NDS Country”框。

2.3 使用查找转换进行 Upsert

使用查找转换进行 Upsert 的操作步骤如下：
1. 在解决方案资源管理器中，右键单击“SSIS 包”，选择“新建 SSIS 包”。
2. 从工具箱中拖动“数据流任务”到设计界面，双击该框进行编辑。
3. 为了将阶段中的国家表设置为源，从工具箱中拖动“OLE DB 源”到设计界面，将其重命名为“Stage Country”。配置连接指向阶段数据库。为了从源表中检索两列并添加三个额外列，将数据访问模式设置为“SQL 命令”，并将 SQL 命令设置为：

select country_code, country_name,
cast(2 as tinyint) as source_system_code,
getdate() as create_timestamp,
getdate() as update_timestamp
from country

4. 点击“预览”确保 SQL 语句正确且数据存在，然后点击“确定”。
5. 创建一个查找转换来检查行是否已存在于 NDS 中。在工具箱的“数据流转换”部分，找到“查找”图标并拖动到设计界面，将其命名为“Lookup Country Code”。将“Stage Country”的绿色箭头连接到“Lookup Country Code”框，双击该框进行编辑。
6. 配置查找转换以检查国家代码是否存在于 NDS 国家表中。将连接设置为 NDS，选择“使用 SQL 查询的结果”，并将查询设置为“select country_code from country”。点击“预览”确保数据存在。
7. 点击“列”选项卡，确保左侧“可用输入列”中的“country_code”与右侧“可用查找列”中的“country_code”有连线。若没有，则从左侧拖动“country_code”到右侧进行映射，无需勾选“country_code”的框。
8. 点击“高级”选项卡，通过将缓存大小设置为接近查找列的字节大小来调整内存使用。勾选“启用内存限制”框，并将缓存大小设置为 1MB 以最小化内存使用。将 SQL 语句更改为“select country_code from country where country_code = ?”。
9. 点击“确定”关闭“查找转换编辑器”对话框。
10. 为了更新 NDS 中已存在的国家的名称，从工具箱中找到“OLE DB 命令”并拖动到设计界面，将其命名为“Update Existing Row”。将“Lookup Country Code”的绿色箭头连接到“Update Existing Row”框，双击该框进行编辑。
11. 在“连接管理器”列下，将“连接”下拉列表设置为“servername.NDS.ETL”。点击“组件属性”选项卡，将“SqlCommand”设置为：

update country
set country_name = ?,
source_system_code = 2,
create_timestamp = getdate(),
update_timestamp = getdate()
where country_code = ?

12. 在上述 Transact SQL 中，两个问号是参数，第一个问号是“Param_0”，第二个问号是“Param_1”。点击“列映射”选项卡，将“country_code”连接到“Param_1”，将“country_name”连接到“Param_0”，然后点击“确定”关闭编辑器。
13. 为了插入阶段表中不存在于 NDS 的行，从工具箱中找到“SQL Server 目标”并拖动到设计界面，将其命名为“Insert New Rows”。将“Lookup Country Code”的红色箭头连接到该框，在对话框中点击“确定”。
14. 双击“Insert New Rows”框进行编辑，将连接设置为 NDS，将“表”设置为“country”，点击“预览”验证所选的表是否正确。
15. 点击“映射”，配置输入列和目标列之间的映射。
16. 点击“确定”关闭“SQL Server 目标编辑器”对话框。
17. 为了测试包，对 NDS 数据库执行以下 SQL 语句准备数据：

update nds.dbo.country set country_name = 'Andorra1'
where country_code = 'ad'
delete from nds.dbo.country where country_code = 'tc'

18. 查询 NDS 国家表，确保数据更改成功。点击“调试”菜单，选择“开始调试 (F5)”执行包。
19. 检查 NDS 国家表，确保安道尔（Andorra）的数据已更正，特克斯和凯科斯群岛（Turks and Caicos Island）已重新插入表中，并且这两条记录的“create_timestamp”和“update_timestamp”列都设置为当天日期。

2.4 方法性能比较

SQL 方法更简单易用，但哪种方法性能最佳取决于数据大小、查找复杂度以及 ETL 服务器的配置，特别是内存。如果查找表可以缓存到 ETL 服务器的内存中，那么查找和 SCD 方法比 SQL 语句方法更快；如果查找表无法缓存到内存中，SQL 语句方法更快。SQL 语句方法的性能还取决于索引和分区。如果在数据仓库项目中性能至关重要，最好的方法是测试两种方法并比较性能。

3. 数据规范化与加载

3.1 规范化概述

规范化是 NDS 数据填充的关键部分。以商店表为例，在 NDS 中，阶段商店表被规范化为四个表：商店（store）、商店类型（store_type）、地址（address）和电话号码（phone_number）。要填充 NDS 商店表，需要先加载地址、商店类型和电话号码表。

3.2 阶段数据库中的商店表

阶段数据库中的商店表如下所示：
| store_number | store_name | store_type | other columns |
| — | — | — | — |
| 1010 | Paris | Mini Outlet | … |
| 1011 | Lyon | Full Outlet | … |
| 1012 | Toulouse | Full Outlet | … |
| 1013 | Nice | Full Outlet | … |
| 1014 | Strasbourg | Distribution Center | … |
| 1015 | Nantes | Mini Outlet | … |

可以看出，“store_type”列未进行规范化。

3.3 NDS 中的商店表

NDS 中的商店表如下所示：
| store_key | store_number | store_name | store_type_key | other columns |
| — | — | — | — | — |
| 0 | 0 | Unknown | 0 | … |
| 562 | 1010 | Paris | 19 | … |
| 607 | 1011 | Lyon | 18 | … |
| 576 | 1012 | Toulouse | 18 | … |
| 531 | 1013 | Nice | 18 | … |
| 585 | 1014 | Strasbourg | 17 | … |
| 547 | 1015 | Nantes | 19 | … |
| 515 | 2703 | Osaka | 18 | … |

要加载 NDS 商店表，需要从商店类型表中获取“store_type_key”，同理，对于地址和电话号码列，也需要相应的键值。在进行查找之前，如果有新的商店类型、地址或电话号码，需要先加载它们。

3.4 创建 SSIS 包加载商店表

创建 SSIS 包加载商店表的操作步骤如下：
1. 打开 BIDS，创建一个新包，命名为“NDS store”。从工具箱中拖动“Execute SQL 任务”到设计界面，命名为“Insert New Store Types”。
2. 双击该任务进行编辑，将连接设置为 NDS，将 SQL 语句设置为：

insert into nds.dbo.store_type
( store_type, source_system_code,
create_timestamp, update_timestamp )
select
distinct store_type as store_type,
1 as source_system_code,
getdate() as create_timestamp,
getdate() as update_timestamp
from stage.dbo.store s
where not exists
( select * from nds.dbo.store_type t
where t.store_type = s.store_type )

此 SQL 语句用于查找阶段商店表中存在但 NDS 商店类型表中不存在的新商店类型，并将其插入 NDS 商店类型表。
3. 同样，创建一个“Execute SQL 任务”，命名为“Insert New Phone Number”，指向 NDS 数据库，将 SQL 语句设置为：

insert into nds.dbo.phone_number
( phone_number, source_system_code,
create_timestamp, update_timestamp )
select
distinct s.phone_number as phone_number,
1 as source_system_code,
getdate() as create_timestamp,
getdate() as update_timestamp
from stage.dbo.store s
where not exists
( select * from nds.dbo.phone_number p
where p.phone_number = s.phone_number )

此 SQL 语句用于查找阶段中存在但 NDS 中不存在的新电话号码，并将其插入 NDS 电话号码表。
4. 使用“Execute SQL 任务”插入新城市和地址，SQL 语句如下：

-- 插入新城市
insert into city
( city_name, source_system_code,
create_timestamp, update_timestamp )
select
distinct s.city as city_name,
1 as source_system_code,
getdate() as create_timestamp,
getdate() as update_timestamp
from stage.dbo.store s
where not exists
( select * from nds.dbo.city c
where c.city_name = s.city )

-- 插入新地址
insert into nds.dbo.address
( address1, address2, address3, address4,
city_key, post_code, state_key, country_key,
source_system_code, create_timestamp, update_timestamp)
select
distinct s.address1, s.address2, s.address3, s.address4,
c.city_key, s.zipcode, st.state_key, co.country_key,
1 as source_system_code,
getdate() as create_timestamp,
getdate() as update_timestamp
from stage.dbo.store s
left join nds.dbo.city c
on s.city = c.city_name
left join nds.dbo.state st
on case when s.state = 'N/A' then 'Unknown'
else s.state end = st.state_name
left join nds.dbo.country co
on s.country = co.country_code
where not exists
( select * from nds.dbo.address a
where coalesce(a.address1,'') = coalesce(s.address1,'')
and   coalesce(a.address2,'') = coalesce(s.address2,'')
and   coalesce(a.address3,'') = coalesce(s.address3,'')
and   coalesce(a.address4,'') = coalesce(s.address4,'')
and   coalesce(a.city_key,0) = coalesce(c.city_key,0)
and   coalesce(a.post_code,'') = coalesce(s.zipcode,'')
and   coalesce(a.state_key,0) = coalesce(st.state_key,0)
and   coalesce(a.country_key,0) = coalesce(co.country_key,0) )

在上述 SQL 语句中，与状态表的连接使用了 case 语句，目的是将源数据中的“N/A”映射到未知记录。地址表中唯一记录的识别依赖于八个列，因此在“where”子句中有八个条件。
5. 创建“Execute SQL 任务”用于商店表，SQL 语句如下：

insert into nds.dbo.store
( store_number, store_name, store_type_key,
address_key, phone_number_key, web_site, region_key,
source_system_code, create_timestamp, update_timestamp )
select
s.store_number as store_number,
s.store_name as store_name,
t.store_type_key as store_type_key,
z.address_key as address_key,
p.phone_number_key as phone_number_key,
s.web_site as web_site,
r.region_key as region_key,
1 as source_system_code,
getdate() as created_timestamp,
getdate() as update_timestamp
from stage.dbo.store s
inner join nds.dbo.store_type t
on s.store_type = t.store_type
inner join
( select a.address_key, a.address1, a.address2, a.address3,
a.address4, a.post_code as zipcode, c.city_name as city,
st.state_name as state, upper(co.country_code) as country
from address a
inner join city c on a.city_key = c.city_key
inner join state st on a.state_key = st.state_key
inner join country co on a.country_key = co.country_key
) z
on coalesce(s.address1,'') = coalesce(z.address1,'')
and coalesce(s.address2,'') = coalesce(z.address2,'')
and coalesce(s.address3,'') = coalesce(z.address3,'')
and coalesce(s.address4,'') = coalesce(z.address4,'')
and coalesce(s.zipcode,'') = coalesce(z.zipcode,'')
and coalesce(s.city,'') = coalesce(z.city,'')
and coalesce(case when s.state = 'N/A' then 'Unknown'
else s.state end,'') = coalesce(z.state,'')
and coalesce(s.country,'') = coalesce(z.country,'')
inner join nds.dbo.phone_number p
on s.phone_number = p.phone_number
inner join nds.dbo.region r
on s.region = r.region_name
where not exists
( select * from nds.dbo.store n
where n.store_number = s.store_number )

此 SQL 语句用于查找 NDS 中不存在的新商店，将其商店类型、地址、电话号码和区域转换为键值，然后插入 NDS 商店表。
6. 创建另一个“Execute SQL 任务”，命名为“Update Store”，用于更新 NDS 商店表中已存在且值发生变化的记录，SQL 语句如下：

update nds.dbo.store
set store_name = s.store_name,
store_type_key = st.store_type_key,
address_key = a.address_key,
phone_number_key = pn.phone_number_key,
region_key = r.region_key,
web_site = s.web_site,
update_timestamp = getdate()
from stage.dbo.store s
inner join nds.dbo.store n
on s.store_number = n.store_number
inner join nds.dbo.store_type st
on n.store_type_key = st.store_type_key
inner join nds.dbo.address a
on n.address_key = a.address_key
inner join nds.dbo.phone_number pn
on n.phone_number_key = pn.phone_number_key
inner join nds.dbo.region r
on n.region_key = r.region_key
inner join nds.dbo.city c
on a.city_key = c.city_key
inner join nds.dbo.state sta
on a.state_key = sta.state_key
inner join nds.dbo.country co
on a.country_key = co.country_key
where coalesce(s.store_name,'') <> coalesce(n.store_name,'')
or coalesce(s.store_type,'') <> coalesce(st.store_type,'')
or coalesce(s.address1,'') <> coalesce(a.address1,'')
or coalesce(s.address2,'') <> coalesce(a.address2,'')
or coalesce(s.address3,'') <> coalesce(a.address3,'')
or coalesce(s.address4,'') <> coalesce(a.address4,'')
or coalesce(a.city_key,0) <> coalesce(c.city_key,0)
or coalesce(a.post_code,'') <> coalesce(s.zipcode,'')
or coalesce(a.state_key,0) <> coalesce(sta.state_key,0)
or coalesce(a.country_key,0) <> coalesce(co.country_key,0)
or coalesce(s.phone_number,'') <> coalesce(pn.phone_number,'')
or coalesce(s.web_site,'') <> coalesce(n.web_site,'')

此 SQL 语句用于更新 NDS 中与阶段表中商店名称、商店类型、地址、电话号码或网站不同的商店记录。
7. 保存并运行该 SSIS 包，其控制流如图所示。
8. 查询五个 NDS 表（城市、地址、电话号码、商店类型和商店），通过与源系统进行比较，确保它们已正确填充。

3.5 规范化注意事项

在比较可空列时，使用“coalesce”函数将 NULL 替换为空字符串或 0，因为 NULL 不能与 NULL 进行比较。例如，应使用“coalesce(a.post_code,’‘) = coalesce(s.zipcode,’‘)”而不是“a.post_code = s.zipcode”。

通过以上步骤，我们可以完成数据的加载、Upsert 操作以及规范化处理，确保数据仓库中的数据准确、规范。在实际应用中，根据具体情况选择合适的方法和配置，以达到最佳性能。

4. 操作流程总结与可视化

4.1 整体操作流程梳理

为了更清晰地理解整个数据处理过程，我们将上述操作步骤进行总结梳理，形成以下操作流程：
1. 数据加载验证 ：查询阶段表和 NDS 状态表，确保数据正确加载到 NDS。
2. 数据更新测试 ：更新阶段数据库数据，运行 SSIS 包，验证 NDS 数据更新情况。
3. Upsert 操作
- SQL 语句方式 ：创建 SSIS 包，设置 Execute SQL 任务，执行更新和插入操作。
- 查找转换方式 ：创建 SSIS 包，配置数据流任务、OLE DB 源、查找转换、OLE DB 命令和 SQL Server 目标，进行数据的更新和插入。
4. 数据规范化加载
- 加载新的商店类型、电话号码、城市和地址到 NDS 对应表。
- 插入新的商店记录到 NDS 商店表。
- 更新 NDS 商店表中已存在且值发生变化的记录。

4.2 操作流程可视化

以下是使用 mermaid 绘制的流程图，展示了整个数据处理的操作流程：

graph LR
    classDef process fill:#E5F6FF,stroke:#73A6FF,stroke-width:2px;

    A(数据加载验证):::process --> B(数据更新测试):::process
    B --> C{Upsert 操作方式}:::process
    C -->|SQL 语句| D(SQL 语句 Upsert):::process
    C -->|查找转换| E(查找转换 Upsert):::process
    D --> F(数据规范化加载):::process
    E --> F
    F --> G(查询验证 NDS 表):::process

4.3 关键操作步骤对比

为了更直观地比较 SQL 语句和查找转换两种 Upsert 方法的操作步骤，我们整理了以下表格：
| 操作步骤 | SQL 语句方式 | 查找转换方式 |
| — | — | — |
| 1 | 创建 SSIS 包，重命名为“NDS country using SQL” | 右键单击“SSIS 包”，新建 SSIS 包 |
| 2 | 拖动两个 Execute SQL 任务，分别命名为“Update NDS Country”和“Insert NDS Country” | 拖动“数据流任务”到设计界面并编辑 |
| 3 | 设置任务连接为“servername.NDS.ETL”，设置 SQL 语句 | 配置“OLE DB 源”，设置 SQL 命令 |
| 4 | 连接两个任务 | 创建“查找转换”，配置连接和查询 |
| 5 | 执行任务 | 配置“OLE DB 命令”和“SQL Server 目标”，设置映射 |
| 6 | - | 准备测试数据，执行包并验证结果 |

5. 常见问题及解决方法

5.1 NULL 值比较问题

在数据处理过程中，比较可空列时经常会遇到 NULL 值比较的问题。由于 NULL 不能与 NULL 进行比较，直接使用“a.post_code = s.zipcode”可能会导致比较失败。解决方法是使用“coalesce”函数将 NULL 替换为空字符串或 0，例如“coalesce(a.post_code,’‘) = coalesce(s.zipcode,’‘)”。

5.2 数据重复问题

在插入新数据时，可能会出现数据重复的情况。例如，在地址表中，如果不进行正确的映射和判断，可能会出现多个“未知”记录。解决方法是在 SQL 语句的“where”子句中设置多个条件，确保唯一记录的识别。如在插入地址记录时，使用八个条件来判断记录是否已存在于 NDS 中。

5.3 性能问题

不同的操作方法在不同的数据规模和服务器配置下可能会有不同的性能表现。如果查找表可以缓存到 ETL 服务器的内存中，查找和 SCD 方法比 SQL 语句方法更快；如果查找表无法缓存到内存中，SQL 语句方法更快。同时，SQL 语句方法的性能还取决于索引和分区。解决方法是在性能要求较高的项目中，测试两种方法并比较性能，选择更合适的方法。

6. 总结与建议

6.1 总结

本文详细介绍了数据仓库中数据加载、Upsert 操作和规范化处理的方法和步骤。通过查询验证数据加载情况，使用 SQL 语句和查找转换进行 Upsert 操作，以及对阶段数据进行规范化处理并加载到 NDS 表中，确保了数据仓库中的数据准确、规范。同时，还讨论了常见问题及解决方法，帮助读者更好地应对实际应用中可能遇到的问题。

6.2 建议

• 在进行数据处理前，充分了解数据的特点和业务需求，选择合适的操作方法。例如，如果数据量较小且查找表可以缓存到内存中，优先考虑查找转换方法；如果数据量较大且查找表无法缓存到内存中，使用 SQL 语句方法可能更合适。
• 在编写 SQL 语句时，注意处理 NULL 值比较和数据重复问题，确保数据的准确性。
• 对于性能要求较高的项目，进行性能测试，根据测试结果优化操作方法和数据库配置，如调整索引和分区。
• 定期对数据仓库进行维护和监控，及时发现和解决潜在的问题，保证数据仓库的稳定运行。

通过遵循以上建议，读者可以更高效地进行数据仓库的数据处理和管理，为业务决策提供更准确、可靠的数据支持。