ETL：跨具有不同字段类型的不同源进行连接

Question

对于跨多个源执行 ETL 的应用程序，在“值”可能相同但表示方式可能不同的情况下，如何处理连接。例如，让我们假设以下虚构的场景：

来源1

• ProductID: 909 // 一个整数
• 产品："Soda"

来源2

• ProductID: "909" // 一个字符串
• 动作类型："Click"

---

假设这些来自我无权修改的两个不同来源（例如，一个可能是 Salesforce 数据，另一个可能是公司数据库）。 ETL 应用程序如何处理字段类型可能以不同方式存储的连接？

Answer 1

我在许多遗留数据库中都看到过这个问题。表位于不同来源的事实并不相关，因为我已经看到这种情况发生在同一个架构、不同的架构以及不同的数据库中。

这个问题有两个方面：可行性和性能。

可行性

我知道的所有数据库都支持数据类型转换和转换。他们中的一些人在幕后默默地做这件事，他们往往做错了。例如，甲骨文在这方面是臭名昭著的，因为它往往会朝着错误的方向转变。我建议始终明确地这样做。

例如（PostgreSQL）：

create table a (product_id int, name varchar(10));
                                             
insert into a (product_id, name) values (909, 'soda');
                                             
create table b (product_id varchar(10), action_type varchar(10));
                                             
insert into b (product_id, action_type) values ('909', 'click');

在 PostgreSQL 中，以下三个查询有效，它们产生相同的结果集（性能是另一回事）：

select * from a join b on a.product_id = b.product_id; -- don't do this

select * from a join b on a.product_id = cast(b.product_id as int);

select * from a join b on cast(a.product_id as varchar) = b.product_id;

如果您键入第一个选项，引擎会在您不知情的情况下默默地将其转换为第二个或第三个查询。这可能会产生您可能无法正确解决的意外/不必要的错误。同样，始终进行显式转换。

性能

在发挥性能时，重要的是要确定哪张桌子是驾驶桌，哪一张是辅助桌。

如果您决定 a 将成为驾驶台，那么您可能会做两件事：

• 在b 一侧转换，如：

   select * from a join b on a.product_id = cast(b.product_id as int);
  

• 通过（如果可以的话）在cast(b.product_id as int) 上创建表达式索引（或索引虚拟索引）来进一步加快查询速度，如下所示：

   create index ix1 on b ((cast(b.product_id as int)));
  

另一方面，如果您决定 b 将成为驾驶台，那么您可能会这样做：

• 在a 一侧转换，如：

   select * from a join b on cast(a.product_id as varchar) = b.product_id;
  

• 或者，通过（如果可以的话）在cast(a.product_id as varchar) 上创建表达式索引（或索引虚拟索引）来进一步加快查询速度，如下所示：

   create index ix2 on a ((cast(a.product_id as varchar))));
  

要决定哪个选项更好，您需要获得两者的执行计划，阅读估计成本并做出决定。有时估计的成本并不那么可靠：它们只是估计，而不是真实的。在危急情况下，我最终会运行这两个选项进行比较。

Answer 2

1. 从各个来源读取数据
2. 转换为相同的数据类型
3. 加入

全部在您选择的 ETL 工具中。问题在哪里，因为我显然错过了它？...

在Informatica 中实现它的方法如下。

Answer 3

我将采用以下方法：

• 创建识别列类型的功能
• 将列类型映射到文本表示
• 强制转换为保存值所需的最大文本长度
• 加入数据

我在一个数据库的上下文中具有相同的场景并应用这种方法。这是因为从数据库中查询动态列集并对它们执行各种 SQL 操作的功能。

其中一位运营商是UNPIVOT。 T-SQL 语句构建是这样的：

SELECT *
FROM 
(
     SELCET RowID
           ,Col001
           ,Col002
           ...
           ,Col00X
     FROM ...
) DS
UNPIVOT
(
  [value] FOR [column] IN ([Col001], [Col002], ... , [Col00X])
) UNPVT;

问题是UNPIVOT IN 子句中的所有列必须具有相同的类型。当然，最简单的解决方法是将CAST/CONVERT 的所有列都设置为NVARCHAR(MAX)，因为它几乎可以存储所有内容，而我们确实做到了，但是查询的执行时间很长。

因此，最好将最大类型保存在 VARCHAR(X) 或 NVARCHNAR(X) 中以将列转换为它。我编写了一个简单的SQL CLR Aggregate 函数，因为我在表中有所有列名和类型，并且希望快速清晰地获得目标类型。它看起来像这样：

using System;
using System.Data;
using System.Data.Sql;
using Microsoft.SqlServer.Server;
using System.Data.SqlTypes;
using System.Collections.Generic;
using System.Text;
using System.IO;
using System.Linq;

[Serializable]
[
    Microsoft.SqlServer.Server.SqlUserDefinedAggregate
    (
        Microsoft.SqlServer.Server.Format.UserDefined,
        IsInvariantToNulls = true,
        IsInvariantToDuplicates = false,
        IsInvariantToOrder = false,
        MaxByteSize = -1
    )
]
/// <summary>
/// Returns the data type with highest precedence. The date types comes in the "[system_type_name]" format.
/// </summary>
public class AnalysisCustomRollupsGetHighestDataTypeConverstionValue : Microsoft.SqlServer.Server.IBinarySerialize
{
    private Dictionary<string, KeyValuePair<string, int>> dataTypesMapping;
    private List<KeyValuePair<string, int>> destinationDataTypes;
    private String[] dataTypesWithoutPredifinedLength;

    public void Init()
    {
        // for the following data types the length is extracting from the input value
        dataTypesWithoutPredifinedLength = new string[] {"nvarchar", "nchar", "varchar", "char"};

        // each data type is mapped to its string corresponding value ("-1" is "MAX", "0" is defined by source)
        dataTypesMapping = new Dictionary<string, KeyValuePair<string, int>> {
                                                                                {"user-defined data types", new KeyValuePair<string, int> ("NVARCHAR", -1)},
                                                                                {"sql_variant", new KeyValuePair<string, int> ("NVARCHAR", -1)},
                                                                                {"xml", new KeyValuePair<string, int> ("NVARCHAR", -1)},
                                                                                {"datetimeoffset", new KeyValuePair<string, int> ("VARCHAR", 34)},
                                                                                {"datetime2", new KeyValuePair<string, int> ("VARCHAR", 27)},
                                                                                {"datetime", new KeyValuePair<string, int> ("VARCHAR", 19)},
                                                                                {"smalldatetime", new KeyValuePair<string, int> ("VARCHAR", 19)},
                                                                                {"date", new KeyValuePair<string, int> ("VARCHAR", 19)},
                                                                                {"time", new KeyValuePair<string, int> ("VARCHAR", 16)},
                                                                                {"float", new KeyValuePair<string, int> ("VARCHAR", 48)},
                                                                                {"real", new KeyValuePair<string, int> ("VARCHAR", 48)},
                                                                                {"decimal", new KeyValuePair<string, int> ("VARCHAR", 48)},
                                                                                {"money", new KeyValuePair<string, int> ("VARCHAR", 19)},
                                                                                {"smallmoney", new KeyValuePair<string, int> ("VARCHAR", 10)},
                                                                                {"bigint", new KeyValuePair<string, int> ("VARCHAR", 26)},
                                                                                {"int", new KeyValuePair<string, int> ("VARCHAR", 14)},
                                                                                {"smallint", new KeyValuePair<string, int> ("VARCHAR", 7)},
                                                                                {"tinyint", new KeyValuePair<string, int> ("VARCHAR", 3)},
                                                                                {"bit", new KeyValuePair<string, int> ("VARCHAR", 1)},
                                                                                {"ntext", new KeyValuePair<string, int> ("NVARCHAR", -1)},
                                                                                {"text", new KeyValuePair<string, int> ("VARCHAR", -1)},
                                                                                {"image", new KeyValuePair<string, int> ("VARCHAR", -1)},
                                                                                {"timestamp", new KeyValuePair<string, int> ("VARCHAR", 8)},
                                                                                {"uniqueidentifier", new KeyValuePair<string, int> ("VARCHAR", 36)},
                                                                                {"nvarchar", new KeyValuePair<string, int> ("NVARCHAR", 0)},
                                                                                {"nchar", new KeyValuePair<string, int> ("NVARCHAR", 0)},
                                                                                {"varchar", new KeyValuePair<string, int> ("VARCHAR", 0)},
                                                                                {"char", new KeyValuePair<string, int> ("VARCHAR", 0)},
                                                                                {"varbinary", new KeyValuePair<string, int> ("NVARCHAR", -1)},
                                                                                {"binary", new KeyValuePair<string, int> ("NVARCHAR", -1)}
                                                                            };

        destinationDataTypes = new List<KeyValuePair<string, int>>();

    }

    public void Accumulate(SqlString value)
    {
        string[] buffer;
        string currentDataTypeName;
        int currentDataLength;
        
        if (value.IsNull)
        {
            return;
        }

        buffer = value.Value.Split('(', ')');
     
        currentDataTypeName = buffer[0].ToLower();
        
        // length is extracting from the source value
        if (dataTypesWithoutPredifinedLength.Contains(currentDataTypeName))
        {
            if(buffer[1].ToUpper() == "MAX")
            {
                buffer[1] = "-1";
            }

            Int32.TryParse(buffer[1], out currentDataLength);

            destinationDataTypes.Add(new KeyValuePair<string, int>(currentDataTypeName, currentDataLength));
        }
        // length is predefined
        else
        {
            destinationDataTypes.Add(new KeyValuePair<string, int>(dataTypesMapping[currentDataTypeName].Key, dataTypesMapping[currentDataTypeName].Value));
        }
    }

    public void Merge(AnalysisCustomRollupsGetHighestDataTypeConverstionValue other)
    {
        destinationDataTypes = destinationDataTypes.Union(other.destinationDataTypes).ToList();
    }

    public SqlString Terminate()
    {
        string output;
        string length;
  
        length = (destinationDataTypes.OrderBy(x => x.Value).First().Value == -1 ? "MAX" : destinationDataTypes.OrderByDescending(x => x.Value).First().Value.ToString());
        
        output = (destinationDataTypes.Exists(x => String.Equals(x.Key.ToUpper(), "NVARCHAR")) ? "NVARCHAR" : "VARCHAR") + "(" + length + ")";
      
        return new SqlString(output);
    }

    public void Read(BinaryReader r)
    {
        if (r == null) throw new ArgumentNullException("r");

        int count = r.ReadInt32();
        destinationDataTypes = new List<KeyValuePair<string, int>>(count);

        for (int i = 0; i < count; i++)
        {
            destinationDataTypes.Add(new KeyValuePair<string, int> (r.ReadString(), r.ReadInt32()));
        }
    }

    public void Write(BinaryWriter w)
    {
        if (w == null) throw new ArgumentNullException("w");

        w.Write(destinationDataTypes.Count);
        foreach (KeyValuePair<string, int> record in destinationDataTypes)
        {
            w.Write(record.Key);
            w.Write(record.Value);
        }
    }
}

它允许我这样做：

SELECT [dbo].[AnalysisCustomRollupsGetHighestDataTypeConverstionValue] ([column_type])
FROM 
(
    VALUES ('VARCHAR(5)')
          ,('INT')
          ,('SMALLMONEY')
) DS ([column_type]);

返回VARCHAR(14)。

我想在您的 ETL 过程中这将更容易实现。

更难的是处理日期。在我的上下文中，所有日期都以这种格式YYYY-MM-DDTHH-MM-SS 作为字符串出现。如果您需要加入日期，其中一些以不同格式的字符串形式出现，例如 2010 5th May 或 something crazy enter in input without any validation，您需要先将它们转换为日期，然后再转换为字符串。