.NET与大数据

  • .NET与大数据已关闭评论
  • 67 次浏览
  • A+
所属分类:.NET技术
摘要

当别人做大数据用Java、Python的时候,我使用.NET做大数据、数据挖掘,这确实是值得一说的事。
写的并不全面,但都是实际工作中的内容。


前言

当别人做大数据用Java、Python的时候,我使用.NET做大数据、数据挖掘,这确实是值得一说的事。
写的并不全面,但都是实际工作中的内容。

.NET在大数据项目中,可以做什么?

  1. 写脚本(使用控制台程序+顶级语句)
  2. 写工具(使用Winform)
  3. 写接口、写服务

使用C#写代码的优点是什么?

  1. 静态类型+匿名类型,一次性使用的实体类就用匿名类型,多次或多个地方使用的实体类就用静态类型,静态类型优于Python,匿名类型优于Java。你是不是想说Python也有静态类型?你倒是写啊?!
  2. 代码的可维护性好,这是相对于Python说的,不一定是语言的锅,还有固有的代码组织习惯,静态类型本身就是很好的注释
  3. 性能好,异步并发的代码易编写。
    想起来一个事,就是前同事用Python2做数据挖掘,先用的es,性能差,改用的clickhouse,我就纳闷,es性能差?现在我想我明白了,我看了其中一个挖掘算法,它需要在双层循环遍历中去请求es进行查询,它没有使用异步,也没有使用多线程,那不就是一个线程在查询吗?我们现网es服务器配置这么强这么多,它居然用一个线程去同步请求,能快才怪!实际上一个请求耗时极短,因为es有各种缓存,而查询条件精确可以命中缓存,所以可以并发请求多个es节点。
    那前同事为什么没有使用异步并发或多线程呢?Python2不支持吗?或者Python2支持,但写起来不方便?或者前同事不会写?

使用.NET开发的优点是什么?

其中一个优点是应用程序类型丰富,目前我用到的应用程序类型有:

  1. 控制台
  2. Winform
  3. Web API
  4. Blazor
    你是不是想说Java和Python也可以写控制台、窗体程序、Web API?一个熟悉Ptyhon的程序员,可不一定会写窗体程序,需要一点时间学习,一个做了几年.NET的程序员天然会写Winform,就是拖控件啊。当然,也可能他们不用Windows。
    每一种应用程序类型,都意味着学习成本,而这些我已经会了,时间就省下了(Blazor一开始不会,学习花了一两天)。

.NET与ClickHouse

我写了一个大杂烩脚本项目,里面有很多工程是查询ClickHouse统计分析,代码流程就是读取Excel数据作为查询输入条件,查询ClickHouse统计分析,统计结果导出到Excel。一个统计分析工作任务小半天就完成了。
用的ORM是我自己写的Dapper.LiteSql。没什么人用,可能是功能不强吧。不过很适合我自己的需求,我自己经常用。
比如:

int count = session.CreateSql<XXX>(@"     select count(distinct t.xxx, t.xxx, t.xxx) as cnt     from xxx t ") .Where(t => t.PassTime >= startTime && t.PassTime <= endTime) .Where("t.Name in (" + kkNames + ")") .QuerySingle<int>(); 

再比如:

var query = session.CreateSql<XXX>(@"         select t.xxx, t.xxx, t.xxx         from xxx t     ")     .Where(t => t.PassTime >= firstTime && t.PassTime <= firstTime.AddDays(7).AddSeconds(-1)); query.Where(t => plateList.Skip((page - 1) * pageSize).Take(pageSize).ToList().Contains(t.PlateNo)); var temp = query.ToList(); 

对于统计查询,我经常SQL和Lambda表达式混写,感觉这样非常灵活。
某些情况下,混写比纯Lambda写法,是要清晰的:

List<XXX> list = session.CreateSql<XXX>(@"     select xxx, xxx as xxx, max(xxx) as xxx     from (     select xxx, toDate(xxx) as xxx, xxx, count(*) as xxx     from (     select distinct t.xxx, t.xxx, t.xxx     from xxx t ").Where(t => t.Xxx != "xxx") .Where(t => t.XxxTime >= startTime && t.XxxTime <= endTime) .Where(t => xxxList.Contains(t.Xxx)) .Where(@"(     (formatDateTime(t.xxx_time ,'%H:%M:%S') >= '07:00:00' and formatDateTime(t.xxx_time ,'%H:%M:%S') <= '08:59:59') or     (formatDateTime(t.xxx_time ,'%H:%M:%S') >= '14:00:00' and formatDateTime(t.xxx_time ,'%H:%M:%S') <= '20:59:59') )") .Append(@")") .GroupBy("xxx, xxx, xxx") .Append(@")      group by xxx, xxx ") .QueryList<XXX>(); 

上述代码说明:

  1. group by写了两种写法比较随意
  2. 三层select嵌套,当然主流ORM都能实现,但不一定易编写、易阅读
  3. 我不用针对ClickHouse去实现formatDateTime,也不用实现toDate、max、distinct、count,也不用纠结是count(*)还是count(1),只要实现的功能足够少,BUG就少。

.NET与ElasticSearch

本打算使用Elasticsearch.Net,为什么没有使用?

  1. 学习成本,项目中没有学习时间,虽然造测试数据是本职工作,但写小工具不是本职工作不能耽误太多时间,所以没有学习时间
  2. 我使用HttpClient查询es,这种查询es的方式和kibana中写的查询语句、以及前同事留下的创建索引的文档、模板最接近,方便抄现成的。下面是一个完整的查询es方法:
public async Task<TicketAgg> QueryAgg(string strStartTime, string strEndTime, string idCard) {     Stopwatch sw = Stopwatch.StartNew();      string esUrl = $"http://{esIPs[_rnd.Next(0, esIPs.Length)]}:24100/out_xxx/_search";      var esQueryBody = new     {         size = 0,         query = new         {             @bool = new             {                 must = new dynamic[]                 {                     new                     {                         range = new                         {                             travel_time = new                             {                                 gte = strStartTime,                                 lte = strEndTime,                                 format = "yyyyMMddHHmmss"                             }                         }                     },                     new                     {                         match_phrase = new                         {                             zjhm = idCard                         }                     }                 }             }         },         aggs = new         {             countByZjhm = new             {                 terms = new                 {                     field = "zjhm",                     size = 10000                 }             }         }     };      string esPostData = JsonConvert.SerializeObject(esQueryBody);     Console.WriteLine($"ES请求URL:{esUrl}");     Console.WriteLine($"ES请求参数:{esPostData}");     HttpClient httpClient = HttpClientFactory.GetClient();     HttpContent content = new StringContent(esPostData);     content.Headers.ContentType = new System.Net.Http.Headers.MediaTypeHeaderValue("application/json");     string strEsResult = await (await httpClient.PostAsync(esUrl, content)).Content.ReadAsStringAsync();     var resultObj = new     {         took = 0,         aggregations = new         {             countByZjhm = new             {                 buckets = new[]                 {                     new                     {                         key = "",                         doc_count = 0                     }                 }             }         }     };     var esResult = JsonConvert.DeserializeAnonymousType(strEsResult, resultObj);      TicketAgg agg = new TicketAgg();     agg.IdCard = idCard;     agg.Count = esResult.aggregations.countByZjhm.buckets[0].doc_count;      sw.Stop();     Console.WriteLine($"统计数据,耗时:{sw.Elapsed.TotalSeconds.ToString("0.000")} 秒");      return agg; } 

代码中esQueryBody和resultObj都是一次性使用的,直接用匿名动态类型,而TicketAgg是需要实例化作为返回值给其它方法使用的,所以定义成静态类型。
下面一段代码,生产测试数据用的:

public async Task MockXxxData(string indexName, int count, DateTime startDate, DateTime endDate, string[] departures, string[] destinations, dynamic peoples) {     int days = (int)endDate.Subtract(startDate).TotalDays;      List<Task> taskList = new List<Task>();     for (int i = 0; i < count; i++)     {         DateTime date = startDate.AddDays(_rnd.Next(0, days + 1));         long time = (long)(_rnd.NextDouble() * 3600 * 24);         var people = peoples[_rnd.Next(0, peoples.Length)];          var esRequestBody = new         {             xxx_type = _rnd.Next(1, 4).ToString(),             zjlx = "xxx",             zjhm = people.zjhm,             xm = people.xm,             departure = departures[_rnd.Next(0, departures.Length)],             destination = destinations[_rnd.Next(0, destinations.Length)],             xxx_date = date.ToString("yyyyMMdd"),             xxx_time = date.AddSeconds(time).ToString("yyyyMMddHHmmss"),             xxx_time = date.AddSeconds(time).AddHours(0.5 + _rnd.NextDouble()).ToString("yyyyMMddHHmmss"),             xxx_time = date.AddSeconds(time).AddDays(-2 + _rnd.NextDouble()).ToString("yyyyMMddHHmmss"),             xxx = "",             xxx = ""         };          var task = ServiceFactory.Get<EsWriteService>().Write(indexName, esRequestBody);         taskList.Add(task);     }     await Task.WhenAll(taskList); } 

上述代码说明:

  1. 程序跑起来生产数据,一般会有几十个线程,也就是请求es的并发量是几十
  2. 如果你觉得几十的并发量,还是有点高,可以在调用的Write异步方法中使用Semaphore类限制一下并发量,代码如下:
private Semaphore _sem = new Semaphore(20, 20); //限制异步请求的并发数量  public async Task<bool> Write(string indexName, dynamic esRequestBody) {     _sem.WaitOne();     try     {         Stopwatch sw = new Stopwatch();         sw.Start();          indexName = $"{indexName}-{DateTime.Now.Year}-{DateTime.Now.Month:00}";         string esUrl = $"http://{esIPs[_rnd.Next(0, esIPs.Length)]}:24100/{indexName}/doc";          string esRequestData = JsonConvert.SerializeObject(esRequestBody);         HttpContent content = new StringContent(esRequestData);         content.Headers.ContentType = new System.Net.Http.Headers.MediaTypeHeaderValue("application/json");         HttpClient httpClient = HttpClientFactory.GetClient();         string strEsResult = await (await httpClient.PostAsync(esUrl, content)).Content.ReadAsStringAsync();         var resultObj = new         {             status = 0         };         var esResult = JsonConvert.DeserializeAnonymousType(strEsResult, resultObj);          sw.Stop();         _log?.Info($"【写入ES索引】【{(esResult.status == 0 ? "成功" : "失败")}】耗时:{sw.Elapsed.TotalSeconds:0.000} 秒,索引名称:{indexName},请求URL:{esUrl},请求参数:{esRequestData}");         return esResult.status == 0;     }     catch     {         throw;     }     finally     {         _sem.Release();     } } 

我最近写了哪些工程

  1. 大杂烩脚本工程,包括查询clickhouse统计分析输出Excel、查询MySQL和Oracle、各种小脚本工具
  2. Blazor工程,做了一套简单的增删改查,精力有限,自己测试用,不用手动改数据库了
  3. 数据挖掘服务,主要是Web API和定时任务
  4. Winform工具,用于测试时创建ES索引、生产模拟数据。为什么写这个?因为做数据挖掘,不给数据,只能自己造了。

为什么从这篇博客看起来这个项目只有我一个人在做?没团队?

还有项目经理、产品经理、前端等一共几个人,项目资金投入少,所以不可能有很多人的。

为什么没有使用Python?

我一开始是想使用Python的,但就我用.NET写的这些东西,如果改用Python,没个2、3年经验,写不顺畅。

我用.NET做一个项目,Swagger有了,创建工程时自带的,当然Python的Swagger也是有的,你可以百度"python 从注释自动生成 swagger",之前看到过一个不错的,没保存,一时半会就找不到了。
用Blazor做了简单的配置页面,测试时不用去手动修改数据库了
写了一个Mock工程,生产模拟测试数据,写入速度可以达到6000条/秒(一条数据请求一次,不是批量写入),界面如下:
.NET与大数据

最后

写此博客是为了给.NET正名,在大数据项目中,.NET大有可为。
我写代码没有用到什么特别的技术,看起来很简单,但也不是随便学学就能写,没个3、5年经验,很难写的这么快。
我写代码也没有什么条条框框,可能不规范,但很灵活。
例如,winform程序注入日志工具类怎么写?来不急百度了,就这么写吧,一样每秒6000条的狂写日志,还不卡界面:

public partial class Form1 : Form, ILog {     ...省略      public Form1()     {         InitializeComponent();          ...省略          //注入日志工具类         ServiceFactory.Get<IndexCreationService>().InjectLog(this);         ServiceFactory.Get<EsWriteService>().InjectLog(this);         ServiceFactory.Get<MockDataService>().InjectLog(this);     } }  internal class EsWriteService : ServiceBase {     ...省略     private ILog? _log;     public void InjectLog(ILog log) => _log = log;      public async Task<bool> Write(string indexName, dynamic esRequestBody)     {         ...省略         _log?.Info("xxx");         ...省略     } } 

就目前这些项目、脚本、工具而言,感觉这就是我写的最佳实践。不知道最佳实践,代码也能写,容易写成屎山,要么写的服务三天两头崩。