一、引用nuget包
ClickHouse.Ado
ClickHouse.Net
二、基础封装类
类库封装借鉴了其它博客
public class ClickHouseHelper : IDisposable
{
private readonly string _connectionString = "Compress=False;BufferSize=32768;SocketTimeout=10000;CheckCompressedHash=False;Compressor=lz4;Host=192.168.0.233;Port=9000;Database=default;User=default;Password=";
private ClickHouseConnection _clickHouseConnection;
#region Constructor
public ClickHouseHelper()
{
this.CreateConnection();
}
public ClickHouseHelper(string connectionString) : this()
{
this._connectionString = connectionString;
}
#endregion
public ClickHouseConnection CreateConnection()
{
if (_clickHouseConnection == null)
{
var settings = new ClickHouseConnectionSettings(_connectionString);
var cnn = new ClickHouseConnection(settings);
if (cnn.State != ConnectionState.Open)
{
cnn.Open();
}
_clickHouseConnection = cnn;
}
return _clickHouseConnection;
}
public void ExecuteNoQuery(string sql, CommandType commandType, params ClickHouseParameter[] parameters)
{
try
{
if (_clickHouseConnection == null)
{
this.CreateConnection();
}
var command = _clickHouseConnection.CreateCommand();
command.CommandText = sql;
command.CommandType = commandType;
AttachParameters(command.Parameters, parameters);
command.ExecuteNonQuery();
}
catch (Exception e)
{
this.Dispose();
throw;
}
}
public void ExecuteNoQuery(string sql, params ClickHouseParameter[] parameters)
{
try
{
ExecuteNoQuery(sql, CommandType.Text, parameters);
}
catch (Exception e)
{
this.Dispose();
throw;
}
}
public T ExecuteScalar<T>(string sql, CommandType commandType, params ClickHouseParameter[] parameters)
{
T result;
try
{
if (_clickHouseConnection == null)
{
this.CreateConnection();
}
var command = _clickHouseConnection.CreateCommand();
command.CommandText = sql;
command.CommandType = commandType;
AttachParameters(command.Parameters, parameters);
result = (T)command.ExecuteScalar();
}
catch (Exception e)
{
this.Dispose();
throw;
}
return result;
}
public T ExecuteScalar<T>(string sql, params ClickHouseParameter[] parameters)
{
T result;
try
{
result = ExecuteScalar<T>(sql, CommandType.Text, parameters);
}
catch (Exception e)
{
this.Dispose();
throw;
}
return result;
}
public IDataReader ExecuteReader(string sql, CommandType commandType, params ClickHouseParameter[] parameters)
{
IDataReader result = null;
try
{
if (_clickHouseConnection == null)
{
this.CreateConnection();
}
var command = _clickHouseConnection.CreateCommand();
command.CommandText = sql;
command.CommandType = commandType;
AttachParameters(command.Parameters, parameters);
result = command.ExecuteReader();
}
catch (Exception e)
{
this.Dispose();
throw;
}
return result;
}
public IDataReader ExecuteReader(string sql, params ClickHouseParameter[] parameters)
{
IDataReader result;
try
{
result = ExecuteReader(sql, CommandType.Text, parameters);
}
catch (Exception e)
{
this.Dispose();
throw;
}
return result;
}
/// <summary>
/// 执行sql返回一个DataTable
/// </summary>
/// <param name="sql">sql语句</param>
/// <param name="commandType">命令类型</param>
/// <param name="parameters">sql参数</param>
/// <returns></returns>
public DataTable ExecuteDataTable(string sql, CommandType commandType, params ClickHouseParameter[] parameters)
{
DataTable result = null;
try
{
var dataReader = ExecuteReader(sql, commandType, parameters);
if (dataReader != null)
{
result = DataReaderToDataTable(dataReader);
}
}
catch (Exception e)
{
this.Dispose();
throw;
}
return result;
}
/// <summary>
/// 执行sql返回一个DataTable
/// </summary>
/// <param name="sql">sql语句</param>
/// <param name="parameters">sql参数</param>
/// <returns></returns>
public DataTable ExecuteDataTable(string sql, params ClickHouseParameter[] parameters)
{
DataTable result;
try
{
result = ExecuteDataTable(sql, CommandType.Text, parameters);
}
catch (Exception e)
{
this.Dispose();
throw;
}
return result;
}
/// <summary>
/// 执行sql返回指定类型的List
/// </summary>
/// <typeparam name="T">需要返回的类型</typeparam>
/// <param name="sql">sql语句</param>
/// <param name="commandType">命令类型</param>
/// <param name="parameters">sql参数</param>
/// <returns></returns>
public List<T> ExecuteList<T>(string sql, CommandType commandType, params ClickHouseParameter[] parameters) where T : class
{
List<T> resultList = new List<T>();
try
{
var dataReader = ExecuteReader(sql, commandType, parameters);
if (dataReader != null)
{
resultList = ReaderToList<T>(dataReader);
}
}
catch (Exception e)
{
this.Dispose();
throw;
}
return resultList;
}
/// <summary>
/// 执行sql返回指定类型的List
/// </summary>
/// <typeparam name="T">需要返回的类型</typeparam>
/// <param name="sql">sql语句</param>
/// <param name="parameters">sql参数</param>
/// <returns></returns>
public List<T> ExecuteList<T>(string sql, params ClickHouseParameter[] parameters) where T : class
{
List<T> resultList = new List<T>();
try
{
resultList = ExecuteList<T>(sql, CommandType.Text, parameters);
}
catch (Exception e)
{
this.Dispose();
throw;
}
return resultList;
}
/// <summary>
/// DataTable分页;注：传入的sql请自己增加排序条件
/// </summary>
/// <param name="sql">sql语句</param>
/// <param name="pageindex">页码</param>
/// <param name="pagesize">每页条数</param>
/// <param name="parameters">sql参数</param>
/// <returns>返回总条数和分页后数据</returns>
public (ulong, DataTable) ExecuteDataTableByPagination(string sql, int pageindex, int pagesize, params ClickHouseParameter[] parameters)
{
DataTable result;
ulong totalCount = 0;
try
{
(string countsql, string pagesql) = GetCountAndPageSql(sql, pageindex, pagesize);
result = ExecuteDataTable(pagesql, CommandType.Text, parameters);
totalCount = ExecuteScalar<ulong>(countsql);
}
catch (Exception e)
{
this.Dispose();
throw;
}
return (totalCount, result);
}
/// <summary>
/// List分页;注：传入的sql请自己增加排序条件
/// </summary>
/// <typeparam name="T">需要返回的list类型</typeparam>
/// <param name="sql">sql语句</param>
/// <param name="pageindex">页码</param>
/// <param name="pagesize">每页条数</param>
/// <param name="parameters">sql参数</param>
/// <returns>返回总条数和分页后数据</returns>
public (ulong, List<T>) ExecuteListByPagination<T>(string sql, int pageindex, int pagesize, params ClickHouseParameter[] parameters) where T : class
{
List<T> result;
ulong totalCount = 0;
try
{
(string countsql, string pagesql) = GetCountAndPageSql(sql, pageindex, pagesize);
result = ExecuteList<T>(pagesql, CommandType.Text, parameters);
totalCount = ExecuteScalar<ulong>(countsql);
}
catch (Exception e)
{
this.Dispose();
throw;
}
return (totalCount, result);
}
/// <summary>
/// 批量新增数据;注：单条增加请使用ExecuteNonQuery
/// </summary>
/// <typeparam name="T">数据类型</typeparam>
/// <param name="sourceList">源数据</param>
/// <param name="tbName">需要插入的表名;注：不填默认为类名</param>
public void BulkInsert<T>(List<T> sourceList, string tbName = "") where T : class
{
tbName = string.IsNullOrEmpty(tbName) ? typeof(T).Name : tbName;
try
{
string insertClickHouseSql = $"INSERT INTO {tbName} ({GetColumns<T>()}) VALUES @bulk;";
if (_clickHouseConnection == null)
{
this.CreateConnection();
}
var command = _clickHouseConnection.CreateCommand();
command.CommandText = insertClickHouseSql;
command.Parameters.Add(new ClickHouseParameter
{
ParameterName = "bulk",
Value = List2AList(sourceList)
});
command.ExecuteNonQuery();
}
catch (Exception e)
{
this.Dispose();
throw;
}
}
#region private
private List<dynamic[]> List2AList<T>(List<T> sourceList)
{
List<dynamic[]> result = new List<dynamic[]>();
sourceList.ForEach(u =>
{
var dic = GetColumnsAndValue(u);
result.Add(dic.Select(i => i.Value).ToArray());
});
return result;
}
private string GetColumns<T>()
{
try
{
var dic = GetColumnsAndValue<T>(default(T));
return string.Join(",", dic.Select(u => u.Key).ToArray());
}
catch (Exception e)
{
this.Dispose();
throw;
}
}
private Dictionary<string, object> GetColumnsAndValue<T>(T u)
{
try
{
Dictionary<string, object> dic = new Dictionary<string, object>();
Type t = typeof(T);
if (u != null)
{
t = u.GetType();
}
var columns = t.GetProperties(BindingFlags.Public | BindingFlags.Instance);
foreach (var item in columns)
{
object v = null;
if (u != null)
{
v = item.GetValue(u);
}
dic.TryAdd(item.Name, v);
}
return dic;
}
catch (Exception e)
{
this.Dispose();
throw e;
}
}
private void AttachParameters(ClickHouseParameterCollection parametersCollection, ClickHouseParameter[] parameters)
{
foreach (var item in parameters)
{
parametersCollection.Add(item);
}
}
/// <summary>
/// 将IDataReader转换为DataTable
/// </summary>
/// <param name="reader"></param>
/// <returns></returns>
private static DataTable DataReaderToDataTable(IDataReader reader)
{
DataTable objDataTable = new DataTable("Table");
int intFieldCount = reader.FieldCount;
for (int intCounter = 0; intCounter < intFieldCount; ++intCounter)
{
objDataTable.Columns.Add(reader.GetName(intCounter).ToUpper(), reader.GetFieldType(intCounter));
}
objDataTable.BeginLoadData();
object[] objValues = new object[intFieldCount];
while (reader.NextResult())
{
while (reader.Read())
{
reader.GetValues(objValues);
objDataTable.LoadDataRow(objValues, true);
}
}
reader.Close();
objDataTable.EndLoadData();
return objDataTable;
}
private static T ReaderToModel<T>(IDataReader dr)
{
try
{
using (dr)
{
if (dr.Read())
{
List<string> list = new List<string>(dr.FieldCount);
for (int i = 0; i < dr.FieldCount; i++)
{
list.Add(dr.GetName(i).ToLower());
}
T model = Activator.CreateInstance<T>();
foreach (PropertyInfo pi in model.GetType().GetProperties(BindingFlags.GetProperty | BindingFlags.Public | BindingFlags.Instance))
{
if (list.Contains(pi.Name.ToLower()))
{
if (!IsNullOrDBNull(dr[pi.Name]))
{
pi.SetValue(model, HackType(dr[pi.Name], pi.PropertyType), null);
}
}
}
return model;
}
}
return default(T);
}
catch (Exception ex)
{
throw ex;
}
}
private static List<T> ReaderToList<T>(IDataReader dr)
{
using (dr)
{
List<string> field = new List<string>(dr.FieldCount);
for (int i = 0; i < dr.FieldCount; i++)
{
field.Add(dr.GetName(i).ToLower());
}
List<T> list = new List<T>();
while (dr.NextResult())
{
while (dr.Read())
{
T model = Activator.CreateInstance<T>();
foreach (PropertyInfo property in model.GetType().GetProperties(BindingFlags.GetProperty | BindingFlags.Public | BindingFlags.Instance))
{
if (field.Contains(property.Name.ToLower()))
{
if (!IsNullOrDBNull(dr[property.Name]))
{
property.SetValue(model, HackType(dr[property.Name], property.PropertyType), null);
}
}
}
list.Add(model);
}
}
return list;
}
}
//这个类对可空类型进行判断转换，要不然会报错
private static object HackType(object value, Type conversionType)
{
if (conversionType.IsGenericType && conversionType.GetGenericTypeDefinition().Equals(typeof(Nullable<>)))
{
if (value == null)
return null;
System.ComponentModel.NullableConverter nullableConverter = new System.ComponentModel.NullableConverter(conversionType);
conversionType = nullableConverter.UnderlyingType;
}
return Convert.ChangeType(value, conversionType);
}
private static bool IsNullOrDBNull(object obj)
{
return ((obj is DBNull) || string.IsNullOrEmpty(obj.ToString())) ? true : false;
}
private (string, string) GetCountAndPageSql(string sql, int pageindex, int pagesize)
{
string countSql = $"SELECT COUNT(1) count FROM ({sql}) A";
string pageSql = $"select * from ({sql}) LIMIT {pagesize} OFFSET {(pageindex - 1) * pagesize}";
return (countSql, pageSql);
}
public void Dispose()
{
_clickHouseConnection?.Dispose();
_clickHouseConnection = null;
GC.Collect();
}
#endregion
} 三、注意事项
1、ClickHouse 数据库对应端口为tcp_port中设置的端口。
2、时区问题。Clickhosue中取出来的时候会多8个小时，之前一度怀疑安装时服务器时区不对，但实际上都是正确的，只能手动将时间通过ToLocalTime转成本地时区。
3、批量插数据。批量插数据的时候如果传入一个List的话，对应的类需要增加GetEnumerator方法。例如：
/// <summary>
/// 测试
/// </summary>
public class flink_user_2
{
public long id { get; set; }
public string name { get; set; }
public int age { get; set; }
public string sex { get; set; }
public string phone { get; set; }
/// <summary>
/// 批量插入必须
/// </summary>
/// <returns></returns>
public IEnumerator GetEnumerator()
{
yield return name;
yield return age;
yield return sex;
yield return phone;
}
} 四、使用示例
public class DBContext
{
public void getTest()
{
ClickHouseHelper helper = new ClickHouseHelper();
var db = helper.ExecuteList<Models.flink_user_2>("select * from flink_user_2");
foreach (var item in db)
{
Console.WriteLine(Newtonsoft.Json.JsonConvert.SerializeObject(item));
}
Console.WriteLine($"总数：{db.Count}");
}
public void InserTest()
{
ClickHouseHelper click = new ClickHouseHelper();
List<flink_user_2> flinks = new List<flink_user_2>();
for (int i = 0; i < 100000000; i++)
{
if (flinks.Count > 10000)
{
Console.WriteLine("插入...." + flinks.Count + "条");
click.BulkInsert<flink_user_2>(flinks, "flink_user_2");
flinks.Clear();
}
flinks.Add(new flink_user_2()
{
age = i,
id = i,
name = "测试" + i,
phone = "12345678911",
sex = "男"
});
}
Console.WriteLine("开始进行插入....");
click.BulkInsert<flink_user_2>(flinks, "flink_user_2");
}
}
