SSIS:将记录集写入文件的脚本任务 [英] SSIS: Script task to write recordset to file
问题描述
我正在使用SQL Server Data Tools 2013创建SSIS包.该程序包具有带有完整结果集"选项的执行SQL任务"，用于将查询结果推送到对象"类型的SSIS变量中.
I am using SQL Server Data Tools 2013 to create an SSIS package. This package has an Execute SQL Task with a Full Resultset option to push the query results into an SSIS Variable, of type Object.
我在脚本任务中使用以下命令来获取存储在对象变量中的记录集并将其写入CSV:
I'm using the following in a script task to take a recordset stored in an object variable and write it to a CSV:
Public Sub Main()
Dim fileName As String = Dts.Variables("vFileName").Value.ToString
Dim destinationPath As String = Dts.Variables("vDestinationPath").Value.ToString
Dim destinationPathAndFileName As String = destinationPath + fileName
Dim fileContents As String = ""
Dim oleDB As OleDbDataAdapter = New OleDbDataAdapter()
Dim table As DataTable = New DataTable()
Dim rs As System.Object = Dts.Variables("vResultSet").Value
' Populate DataTable with vResultSet data
oleDB.Fill(table, rs)
' Loop through columns and concatenate with commas
For Each col As DataColumn In table.Columns
fileContents &= col.ColumnName & ","
Next
' Remove final comma from columns string and append line break
fileContents = fileContents.Substring(0, fileContents.Length - 1)
fileContents &= Environment.NewLine
' Loop through rows and concatenate with commas
Dim i As Integer
For Each row As DataRow In table.Rows
For i = 1 To table.Columns.Count
fileContents &= row(i - 1).ToString() & ","
Next
' Remove final comma from row string and append line break
fileContents = fileContents.Substring(0, fileContents.Length - 1)
fileContents &= Environment.NewLine
Next
' Write all text to destination file. If file exists, this step will overwrite it.
System.IO.File.WriteAllText(destinationPathAndFileName, fileContents)
Dts.TaskResult = ScriptResults.Success
End Sub
这可以工作，但是速度很慢，例如将一个14k行数据集写入CSV需要25分钟以上.我无法使用数据流，因为此过程存在于循环中，并且要导出的每个表的元数据都不同.我敢肯定，脚本任务是唯一的选择，但是有没有比遍历数据集的每一行更快的方法了?请让我知道是否可以提供更多信息.
This works, but it's veeeery slow, like 25+ minutes to write a single 14k-row dataset to CSV. I can't use a data flow because this process exists in a loop, and the metadata for each table to be exported is different. I'm pretty sure a script task is the only option, but is there a faster way than looping through each row of the dataset? Please let me know if I can provide more info.
推荐答案
随意翻译成VB.NET.就像我已经为另一个项目编写了这段代码一样，我将您的请求与我的工作方式融为一体
Feel free to translate to VB.NET as you see fit. Seeing as how I already have this code ~ written for a different project, I mashed your request in with how mine works
传入3个SSIS变量:vFileName，vDestinationPath和vResultSet，Main中的代码会将ado记录集转换为DataTable，然后将其添加到DataSet中并传递给Persist方法. Persist的delimiter默认参数为|.
Passing in 3 SSIS variables: vFileName, vDestinationPath and vResultSet, the code in Main will convert the ado recordset into a DataTable which is then added to a DataSet and passed to the Persist method. Persist has a default parameter for delimiter of |.
此实现完全不尝试处理任何极端情况.它不使用限定符对文本列进行转义，不对嵌入的限定符进行转义，对提要中的换行符执行任何操作，并且OleDbDataAdapter的fill方法中的某些操作因二进制数据而失败，等等.
This implementation does not attempt to deal with any of the corner cases, at all. It does not escape text columns with a qualifier, doesn't escape embedded qualifiers, do anything with newlines in the feeds and something in the OleDbDataAdapter's fill method fails with binary data, etc
public void Main()
{
string fileName = Dts.Variables["User::vFileName"].Value.ToString();
DataSet ds = null;
DataTable dt = null;
string outputFolder = Dts.Variables["User::vDestinationPath"].Value.ToString();
string fileMask = string.Empty;
string sheetName = string.Empty;
string outSubFolder = string.Empty;
string message = string.Empty;
bool fireAgain = true;
try
{
ds = new DataSet();
dt = new DataTable();
System.Data.OleDb.OleDbDataAdapter adapter = new System.Data.OleDb.OleDbDataAdapter();
adapter.Fill(dt, Dts.Variables["User::vResultSet"].Value);
string baseFileName = System.IO.Path.GetFileNameWithoutExtension(fileName);
baseFileName = System.IO.Path.GetFileName(fileName);
ds.Tables.Add(dt);
//foreach (DataTable dt in ds.Tables)
{
Persist(ds, fileName, outputFolder);
}
}
catch (Exception ex)
{
Dts.Events.FireInformation(0, "Data Dumper", string.Format("{0}|{1}", "fileName", fileName), string.Empty, 0, ref fireAgain);
Dts.Events.FireInformation(0, "Data Dumper", string.Format("{0}|{1}", "outputFolder", outputFolder), string.Empty, 0, ref fireAgain);
Dts.Events.FireInformation(0, "Data Dumper", string.Format("{0}|{1}", "ExceptionDetails", ex.ToString()), string.Empty, 0, ref fireAgain);
Dts.Events.FireInformation(0, "Data Dumper", string.Format("{0}|{1}", "InnerExceptionDetails", ex.InnerException), string.Empty, 0, ref fireAgain);
}
Dts.TaskResult = (int)ScriptResults.Success;
}
public static void Persist(System.Data.DataSet ds, string originalFileName, string outputFolder, string delimiter = "|")
{
// Enumerate through all the tables in the dataset
// Save it out as sub versions of the
if (ds == null)
{
return;
}
string baseFileName = System.IO.Path.GetFileNameWithoutExtension(originalFileName);
string baseFolder = System.IO.Path.GetDirectoryName(originalFileName);
System.Collections.Generic.List<string> header = null;
foreach (System.Data.DataTable table in ds.Tables)
{
string outFilePath = System.IO.Path.Combine(outputFolder, string.Format("{0}.{1}.csv", baseFileName, table.TableName));
System.Text.Encoding e = System.Text.Encoding.Default;
if (table.ExtendedProperties.ContainsKey("Unicode") && (bool)table.ExtendedProperties["Unicode"])
{
e = System.Text.Encoding.Unicode;
}
using (System.IO.StreamWriter file = new System.IO.StreamWriter(System.IO.File.Open(outFilePath, System.IO.FileMode.Create), e))
{
table.ExtendedProperties.Add("Path", outFilePath);
// add header row
header = new System.Collections.Generic.List<string>(table.Columns.Count);
foreach (System.Data.DataColumn item in table.Columns)
{
header.Add(item.ColumnName);
}
file.WriteLine(string.Join(delimiter, header));
foreach (System.Data.DataRow row in table.Rows)
{
// TODO: For string based fields, capture the max length
IEnumerable<string> fields = (row.ItemArray).Select(field => field.ToString());
file.WriteLine(string.Join(delimiter, fields));
}
}
}
}
需要运行，但Biml实现看起来像
Need to run but a Biml implementation looks like
<Biml xmlns="http://schemas.varigence.com/biml.xsd">
<Connections>
<OleDbConnection Name="tempdb" ConnectionString="Data Source=localhost\dev2014;Initial Catalog=AdventureWorksDW2014;Provider=SQLNCLI11.0;Integrated Security=SSPI;"/>
</Connections>
<Packages>
<Package Name="so_37059747" ConstraintMode="Linear">
<Variables>
<Variable DataType="String" Name="QuerySource"><![CDATA[SELECT
S.name
, T.name
FROM
sys.schemas AS S
INNER JOIN
sys.tables AS T
ON T.schema_id = S.schema_id;]]></Variable>
<Variable DataType="String" Name="SchemaName">dbo</Variable>
<Variable DataType="String" Name="TableName">foo</Variable>
<Variable DataType="String" Name="QueryTableDump" EvaluateAsExpression="true">"SELECT X.* FROM [" + @[User::SchemaName] + "].[" + @[User::TableName] + "] AS X;"</Variable>
<Variable DataType="Object" Name="rsTables"></Variable>
<Variable DataType="Object" Name="vResultSet"></Variable>
<Variable DataType="String" Name="vFileName" EvaluateAsExpression="true">@[User::SchemaName] + "_" + @[User::TableName] + ".txt"</Variable>
<Variable DataType="String" Name="vDestinationPath">c:\ssisdata\so\Output</Variable>
</Variables>
<Tasks>
<ExecuteSQL
ConnectionName="tempdb"
Name="SQL Generate Loop data"
ResultSet="Full">
<VariableInput VariableName="User.QuerySource" />
<Results>
<Result VariableName="User.rsTables" Name="0" />
</Results>
</ExecuteSQL>
<ForEachAdoLoop SourceVariableName="User.rsTables" Name="FELC Shred rs" ConstraintMode="Linear">
<VariableMappings>
<VariableMapping VariableName="User.SchemaName" Name="0" />
<VariableMapping VariableName="User.TableName" Name="1" />
</VariableMappings>
<Tasks>
<ExecuteSQL
ConnectionName="tempdb"
Name="SQL Generate Export data"
ResultSet="Full">
<VariableInput VariableName="User.QueryTableDump" />
<Results>
<Result VariableName="User.vResultSet" Name="0" />
</Results>
</ExecuteSQL>
<Script ProjectCoreName="ST_RS2CSV" Name="SCR Convert to text">
<ScriptTaskProjectReference ScriptTaskProjectName="ST_RS2CSV" />
</Script>
</Tasks>
</ForEachAdoLoop>
</Tasks>
</Package>
</Packages>
<ScriptProjects>
<ScriptTaskProject ProjectCoreName="ST_RS2CSV" Name="ST_RS2CSV" VstaMajorVersion="0">
<ReadOnlyVariables>
<Variable Namespace="User" VariableName="vFileName" DataType="String" />
<Variable Namespace="User" VariableName="vDestinationPath" DataType="String" />
<Variable Namespace="User" VariableName="vResultSet" DataType="Object" />
</ReadOnlyVariables>
<Files>
<File Path="ScriptMain.cs" BuildAction="Compile">
<![CDATA[namespace DataDumper
{
using System;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Text;
using System.Windows.Forms;
using System.Xml.Linq;
using Microsoft.SqlServer.Dts.Runtime;
[Microsoft.SqlServer.Dts.Tasks.ScriptTask.SSISScriptTaskEntryPointAttribute]
public partial class ScriptMain : Microsoft.SqlServer.Dts.Tasks.ScriptTask.VSTARTScriptObjectModelBase
{
public void Main()
{
string fileName = Dts.Variables["User::vFileName"].Value.ToString();
DataSet ds = null;
DataTable dt = null;
string outputFolder = Dts.Variables["User::vDestinationPath"].Value.ToString();
string fileMask = string.Empty;
string sheetName = string.Empty;
string outSubFolder = string.Empty;
string message = string.Empty;
bool fireAgain = true;
try
{
ds = new DataSet();
dt = new DataTable();
System.Data.OleDb.OleDbDataAdapter adapter = new System.Data.OleDb.OleDbDataAdapter();
adapter.Fill(dt, Dts.Variables["User::vResultSet"].Value);
string baseFileName = System.IO.Path.GetFileNameWithoutExtension(fileName);
baseFileName = System.IO.Path.GetFileName(fileName);
ds.Tables.Add(dt);
//foreach (DataTable dt in ds.Tables)
{
Persist(ds, fileName, outputFolder);
}
}
catch (Exception ex)
{
Dts.Events.FireInformation(0, "Data Dumper", string.Format("{0}|{1}", "fileName", fileName), string.Empty, 0, ref fireAgain);
Dts.Events.FireInformation(0, "Data Dumper", string.Format("{0}|{1}", "outputFolder", outputFolder), string.Empty, 0, ref fireAgain);
Dts.Events.FireInformation(0, "Data Dumper", string.Format("{0}|{1}", "ExceptionDetails", ex.ToString()), string.Empty, 0, ref fireAgain);
Dts.Events.FireInformation(0, "Data Dumper", string.Format("{0}|{1}", "InnerExceptionDetails", ex.InnerException), string.Empty, 0, ref fireAgain);
}
Dts.TaskResult = (int)ScriptResults.Success;
}
public static void Persist(System.Data.DataSet ds, string originalFileName, string outputFolder, string delimiter = "|")
{
// Enumerate through all the tables in the dataset
// Save it out as sub versions of the
if (ds == null)
{
return;
}
string baseFileName = System.IO.Path.GetFileNameWithoutExtension(originalFileName);
string baseFolder = System.IO.Path.GetDirectoryName(originalFileName);
System.Collections.Generic.List<string> header = null;
foreach (System.Data.DataTable table in ds.Tables)
{
string outFilePath = System.IO.Path.Combine(outputFolder, string.Format("{0}.{1}.csv", baseFileName, table.TableName));
System.Text.Encoding e = System.Text.Encoding.Default;
if (table.ExtendedProperties.ContainsKey("Unicode") && (bool)table.ExtendedProperties["Unicode"])
{
e = System.Text.Encoding.Unicode;
}
using (System.IO.StreamWriter file = new System.IO.StreamWriter(System.IO.File.Open(outFilePath, System.IO.FileMode.Create), e))
{
table.ExtendedProperties.Add("Path", outFilePath);
// add header row
header = new System.Collections.Generic.List<string>(table.Columns.Count);
foreach (System.Data.DataColumn item in table.Columns)
{
header.Add(item.ColumnName);
}
file.WriteLine(string.Join(delimiter, header));
foreach (System.Data.DataRow row in table.Rows)
{
// TODO: For string based fields, capture the max length
IEnumerable<string> fields = (row.ItemArray).Select(field => field.ToString());
file.WriteLine(string.Join(delimiter, fields));
}
}
}
}
enum ScriptResults
{
Success = Microsoft.SqlServer.Dts.Runtime.DTSExecResult.Success,
Failure = Microsoft.SqlServer.Dts.Runtime.DTSExecResult.Failure
};
}
}
]]>
</File>
<File Path="Properties\AssemblyInfo.cs" BuildAction="Compile">
using System.Reflection;
using System.Runtime.CompilerServices;
[assembly: AssemblyTitle("AssemblyTitle")]
[assembly: AssemblyDescription("")]
[assembly: AssemblyConfiguration("")]
[assembly: AssemblyCompany("Bill Fellows")]
[assembly: AssemblyProduct("ProductName")]
[assembly: AssemblyCopyright("Copyright @ 2016")]
[assembly: AssemblyTrademark("")]
[assembly: AssemblyCulture("")]
[assembly: AssemblyVersion("1.0.*")]
</File>
</Files>
<AssemblyReferences>
<AssemblyReference AssemblyPath="System" />
<AssemblyReference AssemblyPath="System.Core" />
<AssemblyReference AssemblyPath="System.Data" />
<AssemblyReference AssemblyPath="System.Data.DataSetExtensions" />
<AssemblyReference AssemblyPath="System.Windows.Forms" />
<AssemblyReference AssemblyPath="System.Xml" />
<AssemblyReference AssemblyPath="Microsoft.SqlServer.ManagedDTS.dll" />
<AssemblyReference AssemblyPath="Microsoft.SqlServer.ScriptTask.dll" />
<AssemblyReference AssemblyPath="System.Linq" />
<AssemblyReference AssemblyPath="System.Xml.Linq" />
<AssemblyReference AssemblyPath="Microsoft.VisualBasic" />
</AssemblyReferences>
</ScriptTaskProject>
</ScriptProjects>
</Biml>
这在15秒内丢弃了所有AdventureworksDW2014
That dumped all of AdventureworksDW2014 in 15 seconds
基于此行失败的注释IEnumerable<string> fields = (row.ItemArray).Select(field => field.ToString());
确保您的项目中具有以下using语句.我认为这些扩展位于Linq命名空间中，但可能是Collections
Ensure that you have the following using statements in your project. I think those extensions are in the Linq namespaces but it could have been the Collections
using System;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Text;
using System.Windows.Forms;
using System.Xml.Linq;
using Microsoft.SqlServer.Dts.Runtime;
为什么原来的速度慢?
我的假设是，缓慢归结为所有这些级联.字符串在.Net中是不可变的，并且每次向该字符串添加列时都在创建该字符串的新版本.构建行时，我使用String.Join方法将数组的每个元素压缩为单个字符串.这也简化了附加字段定界符所需的逻辑.
Why was the original slow?
My assumption is the slowness boils down to all that concatenation. Strings are immutable in .Net and you are creating a new version of that string each time you add a column to it. When I build my line, I'm using the String.Join method to zip up each element an array into a single string. This also simplifies the logic required to append the field delimiters.
我还立即将当前行写入文件中，而不是消耗内存以仅通过调用WriteAllText
I also immediately write the current line to a file instead of bloating my memory just to dump it all with a call to WriteAllText
这篇关于SSIS:将记录集写入文件的脚本任务的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
