【问题标题】:Import multiple flat files in the same folder to SQL Server using flat file name as table name使用平面文件名作为表名将同一文件夹中的多个平面文件导入 SQL Server
【发布时间】:2022-01-24 18:18:18
【问题描述】:

我一直在寻找一种通过单击按钮将同一文件夹中的 100 个文本文件上传到 SQL Server 的有效方法。我每个月都执行此操作,手动映射表。在准确提取之前手动选择每个平面文件 (.txt) 并将其映射到正确的加载表需要一些时间。 我搜索了许多不同的方式,包括 SSIS。这些操作过于手动,需要花费太多时间来执行,尤其是维护。

我有一个想法,例如以下同一文件夹中的 5 个平面文件:

  1. Extract_ABC.txt
  2. Extract_YMCA.txt
  3. Extract_WSS.txt
  4. Extract_RMC.txt
  5. Extract_HBO.txt

要以架构 [upload] 及其各自的文件名上传到 SQL 服务器。最终输出将如下所示。

  1. [服务器名称].[上传].Extract_ABC
  2. [服务器名称].[上传].Extract_YMCA
  3. [服务器名称].[上传].Extract_WSS
  4. [服务器名称].[上传].Extract_RMC
  5. [服务器名称].[上传].Extract_HBO

然后可以在查找表中手动为这些表分配位置并插入到正确的表中。除了在服务器中维护位置表之外,这将消除几乎所有手动工作。

请注意,这些表在插入时也必须创建,因为文件在提取时具有唯一的名称。 另外,要注意文件是竖线 ( | ) 分隔的文本文件,而不是 CSV

有谁知道可以执行此操作的方法吗?它必须避免使用 Visual Studio 手动包等。而是将每个平面文件以各自的文件名批量插入到 SQL Server 中。

高效的查询/流程可以为我节省几天时间。 还有一个HUGE,如果它可以在预定时间在服务器端运行,以防止从实际活动的电脑(即 SQL Server 代理等)上传(奖励待定)

【问题讨论】:

  • Powershell 在这里可能非常有用。您可以轻松地遍历文件并使用sqlcmdbcp 将数据导入相关表中。
  • 检查这个答案,情况不同,但我认为它可以帮助您跳出框框思考:stackoverflow.com/questions/54781017/…
  • 附带问题,您的表 真的 都称为 txt 并且位于具有文件(基本)名称的 schema 上吗?如果是这样的话,那是一个糟糕的命名约定......
  • @Larnu 不,这只是一个例子。这些文件实际上是从 SAP 中提取的,并且具有提取和变体编号。这些提取中的每一个都有一个加载和查找表(对原始数据执行的预处理需求)
  • 如果您(现在)也需要CREATE 表格,您怎么知道所述表格的定义是什么?您的文件是否有随附的格式文件? 怎么知道新表中的列应该有哪些数据类型?

标签: sql-server ssis etl flat-file


【解决方案1】:

在这个答案中,我将尝试提供一些信息并提供一个工作示例,将具有不同元数据的多个平面文件导入到单独的 SQL Server 表中。

SSIS 限制

由于 SSIS requires fixed metadata of the data source and destination,我建议使用 C# 代码而不是自动创建 SSIS 包。如果需要 SSIS,您应该使用脚本任务来完成该任务。

在类似的情况下,我一般使用 C# 或其他脚本语言,例如 python。

为什么不应该自动生成 SSIS 包?

将 100 多个平面文件导入应在运行时创建的 100 多个表(没有固定结构)需要创建 100 多个 SSIS 包。即使使用SSIS automation tools 之一自动执行此过程,也很难管理和调试大量 SSIS 包。

统一的 SQL Server 目标表?

如果目标表结构大多具有相似的结构,最好使用架构映射的方式将所有平面文件导入到统一的SQL Server目标中,可以参考以下答案:

解决方案概述

此解决方案中提供的所有代码都可以改进,但它是作为原型编写的

读取平面文件

我创建了一个类TextImport,它尝试遍历给定目录中的所有平面文件,读取平面文件元数据并将每个文件数据存储在System.Data.DataTable 对象中。 (请注意,应添加 Microsoft.VisualBasic 程序集作为引用以使用 TextFieldParser 类)

TextImport txtimp = new TextImport(filename, true, 0); // TextImport(string filename, bool containsHeader, int RowsToSkip)
txtimp.BuildDataTableStructure(); //Read flat file metadata
DataTable dt = txtimp.FillDataTable(); //Convert the flat file to a DataTable object

TextImport 类尝试根据前 8 行检测文件元数据(列数、分隔符、文本限定符、编码)(任意数字)在平面文件中。

创建 SQL 目标表

然后这个System.Data.DataTable 被传递给另一个名为SQLExport 的类,该类根据DataTable 元数据生成并执行CREATE TABLE 语句。

SQLExport.CreateDestinationTable(DataTable dt);

向 SQL 中插入数据

最后提供了两种插入方法:

  1. 生成INSERT INTO () ... VALUES (),(),()... 语句并执行它以将数据插入到 SQL 目标中
  2. 使用System.Data.SqlClient.SqlBulkCopy类通过BULK INSERT方法插入数据。

主要代码应如下所示:

static void Main(string[] args)
{
    //You should set your database connection string
    string connectionstring = @"Data Source=.\SQLINSTANCE;Initial Catalog=tempdb;integrated security=SSPI;";
    //This is the schema of the destination table
    string Schema = "upload";
    //You should set the text files directory
    string directory = @"E:\TextFiles";
    using (SQLExport sqlExp = new SQLExport(connectionstring, Schema))
    {
        //if you don't want to traverse subfolders use System.IO.SearchOption.TopDirectoryOnly
        foreach (string filename in System.IO.Directory.GetFiles(directory,
            "*.txt",System.IO.SearchOption.AllDirectories)){

            using(TextImport txtimp = new TextImport(filename, true, 0))
            {
                txtimp.BuildDataTableStructure();
                DataTable dt = txtimp.FillDataTable();
                dt.TableName = System.IO.Path.GetFileName(filename);
                sqlExp.CreateDestinationTable(dt);
                //Insert using BULK INSERT
                sqlExp.InsertUsingSQLBulk(dt);
                        
                //Creates and Execute an INSERT INTO statment 
                //sqlExp.InsertIntoDb(dt);
            }
        }            
    }

}

测试

我在the following link上存储的两个平面文件上测试了这个解决方案(我的谷歌搜索的第一个结果),如下图所示,表创建成功,数据插入。

请注意,所有列都创建为 NVARCHAR(255),您可以在 SQLExport.cs 文件中更改此设置

您可以在以下 GitHub 存储库中查看完整代码(.NET Framework 4.6.1 C# 控制台应用程序)

如果您决定使用 Script Task 来运行此代码,您应该只将 GitHub 项目中的类复制到您的 Script Task 项目中记得添加 Microsoft.VisualBasic 作为参考。 p>

如何从 SQL Server 代理作业运行 C# 控制台应用程序?

您可以简单地编辑 C# 控制台应用程序以将连接字符串和文本文件目录作为参数传递,然后使用 cmdExec 作业步骤运行它:

甚至使用 SSIS 包:

替代方案 - 使用任务计划程序

另一种方法是使用 Windows 任务计划程序定期运行 C# 控制台应用程序。

【讨论】:

    【解决方案2】:

    正如我所提到的,我实际上会使用 Powershell 之类的工具来执行此操作。执行此操作的脚本实际上非常简单。

    #Below is a Linux Path, as I am running SQL Server (and Powershell) on Linux
    $ImportFolder = "/home/mssql/ImportSample" #The Folder your files are in
    $Instance = "." #The Instance you are inserting the data into. "." means local host
    $Database = "ImportSample" #The database you are inserting the data into
    
    $Files = Get-ChildItem -Path $ImportFolder -Filter "*.txt" #Get all the txt files
    
    #Loop the files
    foreach($File in $Files){
        #Create the statement
        $SQL = "BULK INSERT dbo.[$($File.BaseName.Replace("]","]]"))] FROM '$($File.FullName.Replace("'","''"))' WITH (FORMAT = 'CSV', FIELDTERMINATOR = ',', ROWTERMINATOR = '0x0a', FIRSTROW = 2);"
    
        Write-Debug $SQL
    
        #Execute the statement
        Invoke-SqlCmd -ServerInstance $Instance -Database $Database -Query $SQL
    }
    

    这假定 Windows 身份验证,但是,如果您使用 SQL 身份验证,则可以将 -Username-Password 开关传递给 Invoke-SqlCmd


    注意:这个答案是在CREATE 添加表格的必要性之前写的。这不符合新要求,但是,我将答案留在这里,因为它可能仍然对其他人有类似问题的原始版本的问题有用,没有所述要求。

    【讨论】:

    • 此方法是否适用于管道分隔的文本字段而不是 CSV?
    • 只需为您的终结器@pandasman 适当地更改BULK INSERT 的参数。所以你可以看到我为字段终止符声明了一个逗号(,),为行终止符声明了一个换行符(0x0a)。我确实假设您使用的是 SQL Server 2017+。尽管您不使用引号标识符,但您可以删除旧版本的 FORMAT
    • 我收到一条错误消息:无效的对象名称“实际文件名”(没有 .txt)。这似乎不起作用。我将在下面发布完整的错误。同一文件夹中的两个数据提取显示错误
    • Invoke-SqlCmd:无效的对象名称“dbo.000088_000007”。在 line:15 char:5 + Invoke-SqlCmd -ServerInstance $Instance -Database $Database -Quer ... + ~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + CategoryInfo : InvalidOperation: (:) [Invoke -Sqlcmd],SqlPowerShellSqlExecutionException + FullyQualifiedErrorId:SqlError,Microsoft.SqlServer.Management.PowerShell.GetScriptCommand Invoke-SqlCmd:无效的对象名称'dbo.000089_000003'。在 line:15 char:5 + Invoke-SqlCmd -ServerInstance $Instance -Database $Database -Quer ... +
    • "Invalid object name 'dbo.000088_00000'" 错误告诉你问题所在。该表不存在。 假设这些表存在,并且它暗示在您的问题中。
    【解决方案3】:

    我已将此作为新答案发布,而不是编辑我现有的答案,因为我的原始答案回答了原始问题。

    也就是说,这个答案只是对现有答案的扩展。不幸的是,没有详细说明表中数据的数据类型应该是什么,所以我假设所有数据都是nvarchar(200)。您可以在动态语句中更改它,但请注意,此解决方案假定 所有 数据具有相同的数据类型。如果这不是真的,并且您需要强类型数据,那么是时候提出一个新问题 (don't be a chameleon)。


    无论如何,正如我所说,这扩展了我现有的答案。您会注意到另外一段代码,它动态地处理表的创建。

    我创建了一个临时表(将限制在 Invoke-Sqlcmd 的范围内创建它),到 INSERT 文件中的第一行,我假设具有标题。对于字段终止符,我使用了一个我假设不会出现在您的数据中的字符,一个管道 (|)。如果可以,请使用不同的字符。然后我 INSERT 第一行(因此FIRSTROWLASTROW 但具有1 的值)。这 INSERTs 文件的第一行作为分隔值。

    接下来我创建一个动态语句并将该值拆分为行,然后重新聚合,但引用字段名称,并定义数据类型 (nvarchar(200))。您需要在此处使用尊重序数位置的字符串拆分器,NOT STRING_SPLIT。我使用DelimitedSplitN4K_LEAD(位于我的Utility数据库中),它的定义可以在here找到。

    最后,动态语句被执行,创建了列名按正确顺序排列的表,所有数据类型均为nvarchar(200)Invoke-Sqlcmd 然后执行该语句。

    然后我们回到原来的解决方案,BULK INSERT 数据。

    $ImportFolder = "/home/mssql/ImportSample" #The Folder your files are in
    $Instance = "." #The Instance you are inserting the data into. "." means local host
    $Database = "ImportSample" #The database you are inserting the data into
    
    $Files = Get-ChildItem -Path $ImportFolder -Filter "*.txt" #Get all the txt files
    
    #Loop the files
    foreach($File in $Files){
        #Create the creation statement
        Write-Verbose "Defining table creation script for $($File.Name)."
        $CreateSQL = "CREATE TABLE #Rows (DelimitedData nvarchar(4000));`n" +
                     "BULK INSERT #Rows FROM '$($File.FullName.Replace("'","''"))' WITH(FORMAT = 'CSV', FIELDTERMINATOR = '|', ROWTERMINATOR = '0x0a', FIRSTROW = 1, LASTROW = 1);`n" + 
                     "DECLARE @SQL nvarchar(MAX);`n" +
                     "SELECT @SQL = N'CREATE TABLE dbo.[$($File.BaseName.Replace("]","]]"))] (' + STRING_AGG(QUOTENAME(DSL.Item) + N' nvarchar(200)',N',') WITHIN GROUP (ORDER BY DSL.ItemNumber) + N');' FROM #Rows R CROSS APPLY Utility.fn.DelimitedSplitN4K_LEAD(R.DelimitedData,N',') DSL;`n" + 
                     "EXEC sys.sp_executesql @SQL;"
        
        #Execute the statement
        Write-Verbose "Table creation script for $($File.Name) generated:"
        Write-Verbose $CreateSQL
        Invoke-SqlCmd -ServerInstance $Instance -Database $Database -Query $CreateSQL -Username $User -Password $Password
        
        #Create the Insert statement
        Write-Verbose "Defining BULK INSERT script for $($File.Name)."
        $InsertSQL = "BULK INSERT dbo.[$($File.BaseName.Replace("]","]]"))] FROM '$($File.FullName.Replace("'","''"))' WITH (FORMAT = 'CSV', FIELDTERMINATOR = ',', ROWTERMINATOR = '0x0a', FIRSTROW = 2);"
    
        Write-Verbose "BULK INSERT script for $($File.Name) generated:"
        Write-Verbose $InsertSQL
    
        #Execute the statement
        Invoke-SqlCmd -ServerInstance $Instance -Database $Database -Query $InsertSQL -Username $User -Password $Password
    }
    

    演示此工作的几个屏幕截图(无法用小提琴演示):

    【讨论】:

      【解决方案4】:

      注意:我在底部添加了要在控制台应用程序Core3.1或更高版本中运行的异步代码。

      我有点把这当成一个挑战,看看我能不能做到。这是我在 c# 中提出的...

      基本概念:

      1. 循环文件

      2. 使用管道(|)作为分隔符将每个文件读入数据表

      3. 将标题行读入字符串列表

      4. 根据所有数据类型 varchar(255) 的表头创建表

      5. 使用批量复制将数据表加载到新表中。

         public static string cstr = "Enter your connection string to destination here";
        
         //if you are going to use script task inside SSIS then this is what goes into whatever the Main is in SSIS.
         static void Main(string[] args)
         {
             var dir = @"C:\whereever";
        
             foreach(var file in new DirectoryInfo(dir).GetFiles().AsEnumerable().Where(f=>f.Extension.Contains("txt")))
             {
                 DataTable dt = ConvertCSVtoDataTable(file.FullName, '|');
                 //getColumnNames into an object
                 List<string> colNames = new List<string>();
                 foreach (DataColumn dc in dt.Columns)
                     colNames.Add(dc.ColumnName);
        
                 //create table
                 string tableName = Path.GetFileNameWithoutExtension(file.Name);
                 createTable(colNames, tableName);
                 //load table
                 loadDtIntoDb(dt, colNames, tableName);
             }
         }
        
         public static void loadDtIntoDb(DataTable dt, List<string> cols, string tableName)
         {
             using (var conn = new SqlConnection(cstr))
             {
                 using (var bcp = new SqlBulkCopy(conn))
                 {
                     bcp.DestinationTableName = tableName;
                     foreach (var col in cols)
                         bcp.ColumnMappings.Add(col, col);
                     conn.Open();
                     bcp.WriteToServer(dt);
                 }
             }
         }
        
         public static void createTable(List<string> dcs,string tableName)
         {
             StringBuilder sb = new StringBuilder();
                 sb.AppendLine(string.Format("Create table {0}(", tableName));
             foreach (var dc in dcs)
                 sb.AppendLine(string.Format("{0} varchar(255),", dc));
             //Clean sql by removing final ',' and adding a close paren 
             string sql = sb.ToString().TrimEnd(',') + ")";
             using (var conn = new SqlConnection(cstr))
             {
                 using (var cmd = new SqlCommand(sql, conn))
                 {
                     conn.Open();
                     cmd.ExecuteNonQuery();
                 }
             }
         }
        
         public static DataTable ConvertCSVtoDataTable(string strFilePath, char delim)
         {
             DataTable dt = new DataTable();
             using (StreamReader sr = new StreamReader(strFilePath))
             {
                 string[] headers = sr.ReadLine().Split(delim);
                 foreach (string header in headers)
                 {
                     dt.Columns.Add(header);
                 }
                 while (!sr.EndOfStream)
                 {
                     string[] rows = sr.ReadLine().Split(delim);
                     DataRow dr = dt.NewRow();
                     for (int i = 0; i < headers.Length; i++)
                     {
                         dr[i] = rows[i];
                     }
                     dt.Rows.Add(dr);
                 }
             }
             return dt;
         }
        

      这是我的 using 语句列表:

      using System;
      using System.Collections.Generic;
      using System.Data;
      using System.IO;
      using System.Linq;
      using System.Text;
      using Microsoft.Data.SqlClient;
      

      我看到您正在寻找效率。以下是异步执行的方法:

      public static string cstr = "Enter your connection string to destination here";
      
       //This cannot be done in SSIS
       static async Task Main(string[] args)
       {
           var dir = @"C:\whereever";
       
           var files = new DirectoryInfo(dir).GetFiles().AsEnumerable().Where(f=>f.Extension.Contains("txt")))
           await Task.WhenAll(files.Select(async f=> await processFile(f)));
        }
      
       public statis async Task processFile(File file)
       {
           DataTable dt = await ConvertCSVtoDataTable(file.FullName, '|');
           //getColumnNames into an object
           List<string> colNames = new List<string>();
           foreach (DataColumn dc in dt.Columns)
               colNames.Add(dc.ColumnName);
      
           //create table
           string tableName = Path.GetFileNameWithoutExtension(file.Name);
           await createTable(colNames, tableName);
           //load table
           await loadDtIntoDb(dt, colNames, tableName);
       }
      
      
      
      public static async Task loadDtIntoDb(DataTable dt, List<string> cols, string tableName)
       {
           using (var conn = new SqlConnection(cstr))
           {
               using (var bcp = new SqlBulkCopy(conn))
               {
                   bcp.DestinationTableName = tableName;
                   foreach (var col in cols)
                       bcp.ColumnMappings.Add(col, col);
                   conn.Open();
                   await bcp.WriteToServerAsync(dt);
               }
           }
       }
      
       public static async Task createTable(List<string> dcs,string tableName)
       {
           StringBuilder sb = new StringBuilder();
               sb.AppendLine(string.Format("Create table {0}(", tableName));
           foreach (var dc in dcs)
               sb.AppendLine(string.Format("{0} varchar(255),", dc));
           //Clean sql by removing final ',' and adding a close paren 
           string sql = sb.ToString().TrimEnd(',') + ")";
           using (var conn = new SqlConnection(cstr))
           {
               using (var cmd = new SqlCommand(sql, conn))
               {
                   conn.Open();
                   await cmd.ExecuteNonQueryAsync();
               }
           }
       
      
       public static async Task<DataTable> ConvertCSVtoDataTable(string strFilePath, char delim)
       {
           DataTable dt = new DataTable();
           using (StreamReader sr = new StreamReader(strFilePath))
           {
               string[] headers = await sr.ReadLineAsync().Split(delim);
               foreach (string header in headers)
               {
                   dt.Columns.Add(header);
               }
               while (!sr.EndOfStream)
               {
                   string[] rows = await sr.ReadLineAsync().Split(delim);
                   DataRow dr = dt.NewRow();
                   for (int i = 0; i < headers.Length; i++)
                   {
                       dr[i] = rows[i];
                   }
                   dt.Rows.Add(dr);
               }
           }
           return dt;
       }
      

      【讨论】:

      • 您可以轻松地将其设为异步但不在 SSIS 内部,因此我将其保留为这种格式。
      • 注意:这里写的是 net5.0 Microsoft.Data.SqlClient 是 Nuget,如果使用 SSIS 则需要引用 System.Data.SqlClient。
      【解决方案5】:

      这是一个使用动态脚本的存储过程:

      1. BULK INSERT 将文件放入表[My_bulk_insert_data] 假设文件的第一行包含列名:
      2. 然后在考虑列名的情况下使用动态脚本进行批量插入
      
      create proc usp_bulkinsert_file(@fullfilename as nvarchar(max))
      
      as
      begin
         
      
      declare @filename as nvarchar(max)
      set @filename = substring(@fullfilename,1 + LEN(@fullfilename) - CHARINDEX('\', REVERSE(@fullfilename)) + 1, len(@fullfilename))
      
      declare @script as nvarchar(max)
      if exists(select 1 from sysobjects where name = 'My_bulk_insert_data') drop table My_bulk_insert_data
      set @script = 'if exists(select 1 from sysobjects where name =''' + @filename + ''') drop table [' + @filename + ']'
      exec(@script)
      
      create table My_bulk_insert_data(data nvarchar(max))
      
      set @script='
      BULK INSERT dbo.My_bulk_insert_data
      FROM ''' + @fullfilename + '''
      WITH
      (
         FIELDTERMINATOR = ''''
      ) 
      '
      print @script
      
      exec(@script)
      
      declare @data as nvarchar(max)
      select top 1 @data = data from My_bulk_insert_data
      
      
      
      set @script = 'Create Table [' + @filename + ']('
      select @script = @script + '[' + value + ']' + ' nvarchar(max) ,' from string_split(@data, '|') My_bulk_insert_data
      
      set @script = SUBSTRING(@script, 1, len(@script) - 1) + ')'
      
      print @script
      
      exec(@script)
      
      set @script='BULK INSERT [' + @filename + ']' + ' from ''' + @fullfilename + '''
      with
      (
      FIELDTERMINATOR = ''|''
      )
      ' 
      print @script
      exec(@script)
      
      end
      

      结果截图

      【讨论】:

      • 即使我认为使用脚本语言方法提供了更大的灵活性,我还是喜欢使用动态 SQL 来做到这一点。 +1
      猜你喜欢
      • 2016-07-11
      • 2018-05-26
      • 2019-07-19
      • 1970-01-01
      • 1970-01-01
      • 2013-04-21
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      相关资源
      最近更新 更多