通过 VerticaCopyStream 从客户端进行流式传输

使用 VerticaCopyStream 类可以将数据从客户端系统流式传输到 Vertica 数据库。通过此类，您可以直接使用 SQL COPY 语句，而不必通过将 STDIN 替换为一个或多个数据流先将数据复制到数据库群集中的主机。

注意：

• 使用事务并对 copy 命令禁用自动提交可提高性能。

• 可以通过将 copy 命令与“no commit”修饰符结合使用来禁用自动提交。您必须显式禁用提交。使用 VerticaCopyStream 时，启用事务不会禁用自动提交。

• 与 VerticaCopyStream 结合使用的 copy 命令使用 copy 语法。

• 每次调用 execute 时，都会将 VerticaCopyStream.rejects 置零。如果要捕获拒绝数，请将 VerticaCopyStream.rejects 的值分配给另一个变量，然后再次调用 execute。

• 可以使用 AddStream() 调用添加多个流。

示例用法：

以下示例演示了使用 VerticaCopyStream 将文件流复制到 Vertica。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Data;
using System.IO;
using Vertica.Data.VerticaClient;
namespace ConsoleApplication
{
    class Program
    {
        static void Main(string[] args)
        {
            // Configure connection properties
                    VerticaConnectionStringBuilder builder = new VerticaConnectionStringBuilder();
                builder.Host = "192.168.1.10";
            builder.Database = "VMart";
            builder.User = "dbadmin";
            //open the connection
            VerticaConnection _conn = new VerticaConnection(builder.ToString());
            _conn.Open();
            try
            {
                using (_conn)
                {
                    // Start a transaction
                            VerticaTransaction txn = _conn.BeginTransaction();

                            // Create a table for this example
                            VerticaCommand command = new VerticaCommand("DROP TABLE IF EXISTS copy_table", _conn);
                command.ExecuteNonQuery();
                    command.CommandText = "CREATE TABLE copy_table (Last_Name char(50), "
                                    + "First_Name char(50),Email char(50), "
                                    + "Phone_Number char(15))";
                    command.ExecuteNonQuery();
                    // Create a new filestream from the data file
                            string filename = "C:/customers.txt";
                 Console.WriteLine("\n\nLoading File: " + filename);
                    FileStream inputfile = File.OpenRead(filename);
                    // Define the copy command
                            string copy = "copy copy_table from stdin record terminator E'\n' delimiter '|'" + " enforcelength "
                        + " no commit";
                    // Create a new copy stream instance with the connection and copy statement
                            VerticaCopyStream vcs = new VerticaCopyStream(_conn, copy);

                            // Start the VerticaCopyStream process
                            vcs.Start();
                            // Add the file stream
                            vcs.AddStream(inputfile, false);

                            // Execute the copy
                            vcs.Execute();

                            // Finish stream and write out the list of inserted and rejected rows
                            long rowsInserted = vcs.Finish();
                IList<long> rowsRejected = vcs.Rejects;
                // Does not work when rejected or exceptions defined
                    Console.WriteLine("Number of Rows inserted: " + rowsInserted);
                    Console.WriteLine("Number of Rows rejected: " + rowsRejected.Count);
                    if (rowsRejected.Count > 0)
                    {
                        for (int i = 0; i < rowsRejected.Count; i++)
                        {
                            Console.WriteLine("Rejected row #{0} is row {1}", i, rowsRejected[i]);
                        }
                    }

                            // Commit the changes
                            txn.Commit();
            }
            }
            catch (Exception e)
            {
                Console.WriteLine(e.Message);
            }


                    //close the connection
                    _conn.Close();
    }
    }
}