sql-server - 处理多维立方体的有效方法

Question

我正在使用 SSAS 构建一个多维多维数据集，我根据日期列创建了分区，并为每一天定义了一个分区。源数据大小大于 2 TB。

在部署和处理多维数据集时，如果发生错误，则不会保存所有已处理的分区，并且它们的状态仍未处理。

搜索了一段时间后，我找到了 following article提到:

Parallel (Processing option): Used for batch processing. This setting causes Analysis Services to fork off processing tasks to run in parallel inside a single transaction. If there is a failure, the result is a roll-back of all changes.

搜索后，我找到了一种从 SSIS 包中逐个处理分区的替代方法，如下文所述:

Create SQL Server Analysis Services Partitions using AMO

但处理时间增加了 400% 以上。有没有一种有效的方法可以并行处理分区而不会在发生错误时丢失所有进度？

Answer 1

如果您需要从并行处理选项中受益，那么您不能强制停止所有已处理分区的回滚。

我解决类似问题的首选方法之一是批量处理分区；您可以自动并行处理每个 n 分区，而不是在一次操作中处理所有分区。 (经过多次经验，我发现在我的机器上将 MaxParallel 选项配置为 10 是最佳解决方案)。

那么如果发生错误，只有当前批次会回滚。

在这个答案中，我将尝试提供一个分步指南，以使用 SSIS 自动批量处理分区。

套餐概览

一批建筑尺寸

获取未处理的分区计数

循环分区(每个循环读取 10 个分区)

工艺数据

进程索引

套餐详情

创建变量

首先，我们必须添加一些我们在流程中需要的变量:

intCount, intCurrent: 用于forloop容器

p_Cube:Cube 对象 ID

p_Database:分析数据库 ID

p_MaxParallel:一批要处理的分区数

p_MeasureGroup:度量组对象 ID

p_ServerName:分析服务实例名称 <Machine Name>\<Instance Name>

strProcessData、strProcessDimensions 和 strProcessIndexes:用于存储与处理数据、索引和维度相关的 XMLA 查询

名称以 p_ 开头的所有变量是必需的，可以作为参数添加。

为 Analysis Services 添加连接管理器

添加变量后，我们必须创建一个连接管理器来连接到 SQL Server 分析服务实例:

首先我们必须手动配置连接管理器:

然后我们必须分配服务器名称和初始目录表达式，如下图所示:

将连接管理器重命名为 ssas :

加工尺寸

首先添加一个Sequence Container来隔离包内的维度处理，然后添加一个Script Task和一个Analysis Services Processing Task:





打开脚本任务并选择 p_Database , p_MaxParallel作为只读变量和 strProcessDimensions作为读写变量:



现在，打开脚本编辑器并使用以下代码:

代码是准备 XMLA 命令来处理维度，这个 XMLA 查询将在 Analysis Services 处理任务中使用

#region Namespaces
using System;
using System.Data;
using System.Data.SqlClient;
using Microsoft.SqlServer.Dts.Runtime;
using System.Linq;
using System.Windows.Forms;
using Microsoft.AnalysisServices;
#endregion

namespace ST_00ad89f595124fa7bee9beb04b6ad3d9
{

    [Microsoft.SqlServer.Dts.Tasks.ScriptTask.SSISScriptTaskEntryPointAttribute]
    public partial class ScriptMain : Microsoft.SqlServer.Dts.Tasks.ScriptTask.VSTARTScriptObjectModelBase
    {

        public void Main()
        {
            Server myServer = new Server();

            string ConnStr = Dts.Connections["ssas"].ConnectionString;
            myServer.Connect(ConnStr);

            Database db = myServer.Databases.GetByName(Dts.Variables["p_Database"].Value.ToString());

            int maxparallel = (int)Dts.Variables["p_MaxParallel"].Value;

            var dimensions = db.Dimensions; 

            string strData;

            strData = "<Batch xmlns=\"http://schemas.microsoft.com/analysisservices/2003/engine\"> \r\n <Parallel MaxParallel=\"" + maxparallel.ToString() + "\"> \r\n";

            foreach (Dimension dim in dimensions)
            {
             strData +=
             "    <Process xmlns:xsd=\"http://www.w3.org/2001/XMLSchema\" xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\" xmlns:ddl2=\"http://schemas.microsoft.com/analysisservices/2003/engine/2\" xmlns:ddl2_2=\"http://schemas.microsoft.com/analysisservices/2003/engine/2/2\" xmlns:ddl100_100=\"http://schemas.microsoft.com/analysisservices/2008/engine/100/100\" xmlns:ddl200=\"http://schemas.microsoft.com/analysisservices/2010/engine/200\" xmlns:ddl200_200=\"http://schemas.microsoft.com/analysisservices/2010/engine/200/200\" xmlns:ddl300=\"http://schemas.microsoft.com/analysisservices/2011/engine/300\" xmlns:ddl300_300=\"http://schemas.microsoft.com/analysisservices/2011/engine/300/300\" xmlns:ddl400=\"http://schemas.microsoft.com/analysisservices/2012/engine/400\" xmlns:ddl400_400=\"http://schemas.microsoft.com/analysisservices/2012/engine/400/400\"> \r\n" +
             "     <Object> \r\n" +
             "       <DatabaseID>" + db.ID + "</DatabaseID> \r\n" +
             "       <DimensionID>" + dim.ID + "</DimensionID> \r\n" +
             "     </Object> \r\n" +
             "     <Type>ProcessFull</Type> \r\n" +
             "     <WriteBackTableCreation>UseExisting</WriteBackTableCreation> \r\n" +
             "    </Process> \r\n";
            }

            //}

            strData += " </Parallel> \r\n</Batch>";

            Dts.Variables["strProcessDimensions"].Value = strData;
            Dts.TaskResult = (int)ScriptResults.Success;
        }

        #region ScriptResults declaration

        enum ScriptResults
        {
            Success = Microsoft.SqlServer.Dts.Runtime.DTSExecResult.Success,
            Failure = Microsoft.SqlServer.Dts.Runtime.DTSExecResult.Failure
        };
        #endregion

    }
}

现在，打开 Analysis Services 处理任务并手动定义任何任务，然后转到表达式并分配 strProcessDimensions变量为 ProcessingCommands属性(property):



获取未处理的分区数

为了在块中循环分区，我们首先要获得未处理的分区计数。为此，您必须添加一个脚本任务。选择 p_Cube , p_Database , p_MeasureGroup , p_ServerName变量作为只读变量和 intCount作为读写变量。



在脚本编辑器中编写以下脚本:

#region Namespaces
using System;
using System.Data;
using Microsoft.SqlServer.Dts.Runtime;
using System.Windows.Forms;
using Microsoft.AnalysisServices;
using System.Linq;
#endregion

namespace ST_e3da217e491640eca297900d57f46a85
{

    [Microsoft.SqlServer.Dts.Tasks.ScriptTask.SSISScriptTaskEntryPointAttribute]
    public partial class ScriptMain : Microsoft.SqlServer.Dts.Tasks.ScriptTask.VSTARTScriptObjectModelBase
    {

        public void Main()
        {
            // TODO: Add your code here
            Server myServer = new Server();

            string ConnStr = Dts.Connections["ssas"].ConnectionString;
            myServer.Connect(ConnStr);

            Database db  = myServer.Databases.GetByName(Dts.Variables["p_Database"].Value.ToString());
            Cube objCube = db.Cubes.FindByName(Dts.Variables["p_Cube"].Value.ToString());
            MeasureGroup objMeasureGroup = objCube.MeasureGroups[Dts.Variables["p_MeasureGroup"].Value.ToString()];

            Dts.Variables["intCount"].Value = objMeasureGroup.Partitions.Cast<Partition>().Where(x => x.State != AnalysisState.Processed).Count();

            Dts.TaskResult = (int)ScriptResults.Success;
        }

        #region ScriptResults declaration

        enum ScriptResults
        {
            Success = Microsoft.SqlServer.Dts.Runtime.DTSExecResult.Success,
            Failure = Microsoft.SqlServer.Dts.Runtime.DTSExecResult.Failure
        };
        #endregion

    }
}

以块为单位的进程分区

最后一步是创建一个 Forloop 容器并如下图所示进行配置:

初始表达式:@intCurrent = 0

评估表达式:@intCurrent < @intCount

AssignExpression = @intCurrent + @p_MaxParallel

在 For Loop 容器内添加一个脚本任务来准备 XMLA 查询并添加两个分析服务处理任务，如下图所示:



在脚本任务中，选择 p_Cube , p_Database , p_MaxParallel , p_MeasureGroup作为只读变量，然后选择 strProcessData , strProcessIndexes作为读写变量。



在脚本编辑器中编写以下脚本:

该脚本是准备分别处理分区数据和索引所需的 XMLA 命令

#region Namespaces
using System;
using System.Data;
using System.Data.SqlClient;
using Microsoft.SqlServer.Dts.Runtime;
using System.Linq;
using System.Windows.Forms;
using Microsoft.AnalysisServices;
#endregion

namespace ST_00ad89f595124fa7bee9beb04b6ad3d9
{

    [Microsoft.SqlServer.Dts.Tasks.ScriptTask.SSISScriptTaskEntryPointAttribute]
    public partial class ScriptMain : Microsoft.SqlServer.Dts.Tasks.ScriptTask.VSTARTScriptObjectModelBase
    {


        public void Main()
        {
            Server myServer = new Server();

            string ConnStr = Dts.Connections["ssas"].ConnectionString;
            myServer.Connect(ConnStr);

            Database db = myServer.Databases.GetByName(Dts.Variables["p_Database"].Value.ToString());
            Cube objCube = db.Cubes.FindByName(Dts.Variables["p_Cube"].Value.ToString());
            MeasureGroup objMeasureGroup = objCube.MeasureGroups[Dts.Variables["p_MeasureGroup"].Value.ToString()];
            int maxparallel = (int)Dts.Variables["p_MaxParallel"].Value;


            int intcount = objMeasureGroup.Partitions.Cast<Partition>().Where(x => x.State != AnalysisState.Processed).Count();

            if (intcount > maxparallel)
            {
                intcount = maxparallel;
            }

            var partitions = objMeasureGroup.Partitions.Cast<Partition>().Where(x => x.State != AnalysisState.Processed).OrderBy(y => y.Name).Take(intcount);

            string strData, strIndexes;

            strData = "<Batch xmlns=\"http://schemas.microsoft.com/analysisservices/2003/engine\"> \r\n <Parallel MaxParallel=\"" + maxparallel.ToString() + "\"> \r\n";
            strIndexes = "<Batch xmlns=\"http://schemas.microsoft.com/analysisservices/2003/engine\"> \r\n <Parallel MaxParallel=\"" + maxparallel.ToString() + "\"> \r\n";

            string SQLConnStr = Dts.Variables["User::p_DatabaseConnection"].Value.ToString();



            foreach (Partition prt in partitions)
            {


                strData +=
                 "    <Process xmlns:xsd=\"http://www.w3.org/2001/XMLSchema\" xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\" xmlns:ddl2=\"http://schemas.microsoft.com/analysisservices/2003/engine/2\" xmlns:ddl2_2=\"http://schemas.microsoft.com/analysisservices/2003/engine/2/2\" xmlns:ddl100_100=\"http://schemas.microsoft.com/analysisservices/2008/engine/100/100\" xmlns:ddl200=\"http://schemas.microsoft.com/analysisservices/2010/engine/200\" xmlns:ddl200_200=\"http://schemas.microsoft.com/analysisservices/2010/engine/200/200\" xmlns:ddl300=\"http://schemas.microsoft.com/analysisservices/2011/engine/300\" xmlns:ddl300_300=\"http://schemas.microsoft.com/analysisservices/2011/engine/300/300\" xmlns:ddl400=\"http://schemas.microsoft.com/analysisservices/2012/engine/400\" xmlns:ddl400_400=\"http://schemas.microsoft.com/analysisservices/2012/engine/400/400\"> \r\n " +
                 "      <Object> \r\n " +
                 "        <DatabaseID>" + db.Name + "</DatabaseID> \r\n " +
                 "        <CubeID>" + objCube.ID + "</CubeID> \r\n " +
                 "        <MeasureGroupID>" + objMeasureGroup.ID + "</MeasureGroupID> \r\n " +
                 "        <PartitionID>" + prt.ID + "</PartitionID> \r\n " +
                 "      </Object> \r\n " +
                 "      <Type>ProcessData</Type> \r\n " +
                 "      <WriteBackTableCreation>UseExisting</WriteBackTableCreation> \r\n " +
                 "    </Process> \r\n";

                strIndexes +=
                "    <Process xmlns:xsd=\"http://www.w3.org/2001/XMLSchema\" xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\" xmlns:ddl2=\"http://schemas.microsoft.com/analysisservices/2003/engine/2\" xmlns:ddl2_2=\"http://schemas.microsoft.com/analysisservices/2003/engine/2/2\" xmlns:ddl100_100=\"http://schemas.microsoft.com/analysisservices/2008/engine/100/100\" xmlns:ddl200=\"http://schemas.microsoft.com/analysisservices/2010/engine/200\" xmlns:ddl200_200=\"http://schemas.microsoft.com/analysisservices/2010/engine/200/200\" xmlns:ddl300=\"http://schemas.microsoft.com/analysisservices/2011/engine/300\" xmlns:ddl300_300=\"http://schemas.microsoft.com/analysisservices/2011/engine/300/300\" xmlns:ddl400=\"http://schemas.microsoft.com/analysisservices/2012/engine/400\" xmlns:ddl400_400=\"http://schemas.microsoft.com/analysisservices/2012/engine/400/400\"> \r\n " +
                "      <Object> \r\n " +
                "        <DatabaseID>" + db.Name + "</DatabaseID> \r\n " +
                "        <CubeID>" + objCube.ID + "</CubeID> \r\n " +
                "        <MeasureGroupID>" + objMeasureGroup.ID + "</MeasureGroupID> \r\n " +
                "        <PartitionID>" + prt.ID + "</PartitionID> \r\n " +
                "      </Object> \r\n " +
                "      <Type>ProcessIndexes</Type> \r\n " +
                "      <WriteBackTableCreation>UseExisting</WriteBackTableCreation> \r\n " +
                "    </Process> \r\n";



            }

            strData += " </Parallel> \r\n</Batch>";
            strIndexes += " </Parallel> \r\n</Batch>";

            Dts.Variables["strProcessData"].Value = strData;
            Dts.Variables["strProcessIndexes"].Value = strIndexes;

            Dts.TaskResult = (int)ScriptResults.Success;
        }

        #region ScriptResults declaration

        enum ScriptResults
        {
            Success = Microsoft.SqlServer.Dts.Runtime.DTSExecResult.Success,
            Failure = Microsoft.SqlServer.Dts.Runtime.DTSExecResult.Failure
        };
        #endregion

    }
}

现在打开两个 Analysis Services 处理任务并手动定义任何任务(只是为了验证任务)。然后转到表达式并分配 strProcessData变量为 ProcessingCommands第一个任务中的属性和 strProcessIndexes变量为 ProcessingCommands .



现在您可以执行包，如果发生错误，则只有当前批处理会回滚(10 个分区)。

可能的改进

您可以添加一些日志任务来跟踪包进度，尤其是在处理大量分区时。

由于它包含有用的详细信息，我将此答案发布在我的个人博客上:

SSAS - Efficient way to process a multidimensional cube

我还发表了一篇文章，其中包含有关 SQLShack 的更多详细信息:

An efficient approach to process a SSAS multidimensional OLAP cube