C#에서 Hive 조회·적재하기

// HiveJob.cs
// 대상: .NET Framework 4.5 콘솔 앱 (C# 5 문법만 사용)
// 참조 추가: System.Configuration  (System.Data 는 기본 참조)
// 실행: HiveJob.exe            -> 전일(어제) 파티션 생성
//       HiveJob.exe 20261026   -> 지정한 날짜 파티션 생성
using System;
using System.Configuration;
using System.Data;
using System.Data.Odbc;
using System.Globalization;

namespace HiveJob
{
    /// <summary>
    /// HiveServer2 에 ODBC 로 접속해 조회/실행하는 작은 래퍼.
    /// 호출마다 연결을 열고 닫는다 (ODBC 연결 풀링이 재사용을 처리).
    /// </summary>
    public sealed class HiveClient
    {
        private readonly string _connectionString;
        private readonly int _commandTimeoutSeconds;

        public HiveClient(string connectionString, int commandTimeoutSeconds)
        {
            if (string.IsNullOrEmpty(connectionString))
                throw new ArgumentException("연결 문자열이 비어 있습니다.", "connectionString");

            _connectionString = connectionString;
            _commandTimeoutSeconds = commandTimeoutSeconds;
        }

        /// <summary>SELECT 결과를 DataTable 로 돌려준다. ? 자리에 args 가 순서대로 들어간다.</summary>
        public DataTable Query(string sql, params object[] args)
        {
            using (var conn = new OdbcConnection(_connectionString))
            using (var cmd = CreateCommand(conn, sql, args))
            using (var adapter = new OdbcDataAdapter(cmd))
            {
                var table = new DataTable();
                conn.Open();
                adapter.Fill(table);
                return table;
            }
        }

        /// <summary>INSERT OVERWRITE, CREATE TABLE AS SELECT, SET 등 결과가 없는 문장을 실행한다.</summary>
        public void Execute(string sql, params object[] args)
        {
            using (var conn = new OdbcConnection(_connectionString))
            using (var cmd = CreateCommand(conn, sql, args))
            {
                conn.Open();
                // Hive 드라이버는 영향받은 행 수를 대부분 -1 로 돌려주므로 반환값은 쓰지 않는다.
                cmd.ExecuteNonQuery();
            }
        }

        private OdbcCommand CreateCommand(OdbcConnection conn, string sql, object[] args)
        {
            var cmd = conn.CreateCommand();
            cmd.CommandText = sql;
            cmd.CommandTimeout = _commandTimeoutSeconds;   // 0 이면 무제한

            if (args != null)
            {
                foreach (var value in args)
                {
                    var p = cmd.CreateParameter();
                    p.Value = value ?? DBNull.Value;
                    cmd.Parameters.Add(p);                 // ODBC 는 이름이 아니라 순서로 바인딩
                }
            }
            return cmd;
        }
    }

    internal static class Program
    {
        private static int Main(string[] args)
        {
            try
            {
                var setting = ConfigurationManager.ConnectionStrings["Hive"];
                if (setting == null)
                    throw new ConfigurationErrorsException("App.config 의 connectionStrings 에 'Hive' 항목이 없습니다.");

                int timeout;
                if (!int.TryParse(ConfigurationManager.AppSettings["HiveCommandTimeout"], out timeout))
                    timeout = 3600;

                var hive = new HiveClient(setting.ConnectionString, timeout);

                // 기준일: 인자가 있으면 그 날짜(yyyyMMdd), 없으면 전일
                DateTime baseDate = args.Length > 0
                    ? DateTime.ParseExact(args[0], "yyyyMMdd", CultureInfo.InvariantCulture)
                    : DateTime.Today.AddDays(-1);
                string dt = baseDate.ToString("yyyyMMdd", CultureInfo.InvariantCulture);

                Log("작업 시작 dt=" + dt);

                // 파티션 값은 위에서 DateTime 으로 만든 문자열이라 외부 입력이 섞이지 않는다.
                // (PARTITION 절에는 ? 파라미터를 쓸 수 없어 문자열로 넣는다.)
                string insertSql = string.Format(
                    "INSERT OVERWRITE TABLE mart.daily_sales PARTITION (dt = '{0}') " +
                    "SELECT store_id, SUM(amount) AS total_amount, COUNT(*) AS order_cnt " +
                    "FROM raw.orders " +
                    "WHERE dt = '{0}' " +
                    "GROUP BY store_id", dt);

                hive.Execute(insertSql);
                Log("INSERT OVERWRITE 완료");

                // 일반 조건절 값은 ? 파라미터로 넘긴다.
                DataTable result = hive.Query(
                    "SELECT COUNT(*) AS cnt FROM mart.daily_sales WHERE dt = ?", dt);
                long count = Convert.ToInt64(result.Rows[0][0], CultureInfo.InvariantCulture);
                Log("적재 건수=" + count);

                if (count == 0)
                {
                    Log("적재 건수가 0 건이라 실패로 처리합니다.");
                    return 1;
                }
                return 0;   // 작업 스케줄러에서 0 = 성공
            }
            catch (Exception ex)
            {
                Console.Error.WriteLine(Stamp() + " 오류: " + ex);
                return 1;   // 0 이 아니면 실패로 감지
            }
        }

        private static void Log(string message)
        {
            Console.WriteLine(Stamp() + " " + message);
        }

        private static string Stamp()
        {
            return DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss", CultureInfo.InvariantCulture);
        }
    }
}

<?xml version="1.0" encoding="utf-8"?>
<configuration>
  <startup>
    <supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.5" />
  </startup>
  <connectionStrings>
    <!-- 드라이버 이름은 ODBC 데이터 원본 관리자 > 드라이버 탭에 보이는 이름과 정확히 같아야 합니다. -->
    <add name="Hive"
         connectionString="Driver={Cloudera ODBC Driver for Apache Hive};Host=hive-server01;Port=10000;Schema=default;HiveServerType=2;AuthMech=3;UID=etl_user;PWD=********;" />
    <!-- DSN 을 미리 만들어 둔 경우: connectionString="DSN=HiveProd;UID=etl_user;PWD=********;" -->
  </connectionStrings>
  <appSettings>
    <!-- 명령 타임아웃(초). 대용량 INSERT OVERWRITE 는 넉넉하게. 0 = 무제한 -->
    <add key="HiveCommandTimeout" value="3600" />
  </appSettings>
</configuration>

// Program.cs  (.NET Framework 4.5 콘솔 앱, 추가 참조 없음)
using System;
using System.Data.Odbc;

class Program
{
    static void Main()
    {
        string connStr =
            "Driver={Cloudera ODBC Driver for Apache Hive};" +
            "Host=hive-server01;Port=10000;Schema=default;" +
            "HiveServerType=2;AuthMech=3;UID=etl_user;PWD=********;";

        using (var conn = new OdbcConnection(connStr))
        {
            conn.Open();

            using (var cmd = new OdbcCommand("SELECT * FROM sample_table LIMIT 10", conn))
            using (var reader = cmd.ExecuteReader())
            {
                while (reader.Read())
                {
                    var values = new object[reader.FieldCount];
                    reader.GetValues(values);
                    Console.WriteLine(string.Join("\t", values));
                }
            }
        }

        Console.WriteLine("완료. 아무 키나 누르세요.");
        Console.ReadKey();
    }
}