|
|
|
|
@ -4,8 +4,8 @@ using Microsoft.Extensions.Hosting;
|
|
|
|
|
using Microsoft.Extensions.Logging;
|
|
|
|
|
using System;
|
|
|
|
|
using System.Collections.Generic;
|
|
|
|
|
using System.Diagnostics;
|
|
|
|
|
using System.IO;
|
|
|
|
|
using System.Linq;
|
|
|
|
|
using System.Net.Http;
|
|
|
|
|
using System.Net.Http.Headers;
|
|
|
|
|
using System.Text;
|
|
|
|
|
@ -21,7 +21,7 @@ namespace Kafka2Doris
|
|
|
|
|
private readonly IConfiguration _config;
|
|
|
|
|
private readonly IHttpClientFactory _httpClientFactory;
|
|
|
|
|
|
|
|
|
|
public Worker(ILogger<Worker> logger, IConfiguration config,IHttpClientFactory httpClientFactory)
|
|
|
|
|
public Worker(ILogger<Worker> logger, IConfiguration config, IHttpClientFactory httpClientFactory)
|
|
|
|
|
{
|
|
|
|
|
this._logger = logger;
|
|
|
|
|
this._config = config;
|
|
|
|
|
@ -34,17 +34,17 @@ namespace Kafka2Doris
|
|
|
|
|
{
|
|
|
|
|
_logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
|
|
|
|
|
//
|
|
|
|
|
var topics = _config["topics"].Split(',');
|
|
|
|
|
var topics = _config.GetSection("kafka").GetValue("topics", "").Split(',');
|
|
|
|
|
foreach (var topic in topics)
|
|
|
|
|
{
|
|
|
|
|
var conf = new ConsumerConfig
|
|
|
|
|
{
|
|
|
|
|
BootstrapServers = _config.GetValue("server", "localhost:9092"),
|
|
|
|
|
BootstrapServers = _config.GetSection("kafka").GetValue("host", "localhost:9092"),
|
|
|
|
|
GroupId = $"doris-v1",
|
|
|
|
|
AutoOffsetReset = AutoOffsetReset.Earliest,
|
|
|
|
|
EnableAutoCommit=false
|
|
|
|
|
EnableAutoCommit = false
|
|
|
|
|
};
|
|
|
|
|
var timeout = TimeSpan.FromSeconds(_config.GetValue("timeout",5));
|
|
|
|
|
var timeout = TimeSpan.FromSeconds(_config.GetValue("timeout", 5));
|
|
|
|
|
try
|
|
|
|
|
{
|
|
|
|
|
using (var consumer = new ConsumerBuilder<Ignore, string>(conf).Build())
|
|
|
|
|
@ -53,7 +53,8 @@ namespace Kafka2Doris
|
|
|
|
|
{
|
|
|
|
|
consumer.Subscribe(topic);
|
|
|
|
|
CancellationTokenSource cts = new CancellationTokenSource();
|
|
|
|
|
Console.CancelKeyPress += (_, e) => {
|
|
|
|
|
Console.CancelKeyPress += (_, e) =>
|
|
|
|
|
{
|
|
|
|
|
e.Cancel = true; // prevent the process from terminating.
|
|
|
|
|
cts.Cancel();
|
|
|
|
|
};
|
|
|
|
|
@ -61,12 +62,12 @@ namespace Kafka2Doris
|
|
|
|
|
{
|
|
|
|
|
var max = _config.GetValue("max", 1000);
|
|
|
|
|
var list = new List<string>(max);
|
|
|
|
|
while (max>0)
|
|
|
|
|
while (max > 0)
|
|
|
|
|
{
|
|
|
|
|
try
|
|
|
|
|
{
|
|
|
|
|
var consumeResult = consumer.Consume(timeout);
|
|
|
|
|
if(consumeResult == null)
|
|
|
|
|
if (consumeResult == null)
|
|
|
|
|
{
|
|
|
|
|
break;
|
|
|
|
|
}
|
|
|
|
|
@ -84,32 +85,83 @@ namespace Kafka2Doris
|
|
|
|
|
}
|
|
|
|
|
max -= 1;
|
|
|
|
|
}
|
|
|
|
|
if(list.Count>0)
|
|
|
|
|
if (list.Count > 0)
|
|
|
|
|
{
|
|
|
|
|
var httpClient = this._httpClientFactory.CreateClient();
|
|
|
|
|
var username = _config.GetValue("username", "root");
|
|
|
|
|
var password = _config.GetValue("password", "aA123456!");
|
|
|
|
|
var username = _config.GetSection("doris").GetValue("username", "root");
|
|
|
|
|
var password = _config.GetSection("doris").GetValue("password", "aA123456!");
|
|
|
|
|
httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Basic", Convert.ToBase64String(Encoding.UTF8.GetBytes($"{username}:{password}")));
|
|
|
|
|
httpClient.DefaultRequestHeaders.Add("Expect", "100-continue");
|
|
|
|
|
httpClient.DefaultRequestHeaders.Add("strip_outer_array", "true");
|
|
|
|
|
httpClient.DefaultRequestHeaders.Add("strip_outer_array", "true");
|
|
|
|
|
httpClient.DefaultRequestHeaders.Add("columns", "UserName,SecurityStamp,PasswordHash,PasswordConfirmed,Email,EmailConfirmed,PhoneNumber,PhoneNumberConfirmed,RealName,IdentityNumber,IdentityConfirmed,NickName,Avatar,Sex,Birthday,LockoutEnabled,AccessFailedCount,LockoutEnd,RowVersion,Created,Modified,Deleted");
|
|
|
|
|
httpClient.DefaultRequestHeaders.Add("jsonpaths", "\"$.UserName\", \"$.SecurityStamp\", \"$.PasswordHash`\", \"$.PasswordConfirmed\", \"$.Email\", \"$.EmailConfirmed\", \"$.PhoneNumber\", \"$.PhoneNumberConfirmed\", \"$.RealName\", \"$.IdentityNumber\", \"$.IdentityConfirmed\", \"$.NickName\", \"$.Avatar\", \"$.Sex\", \"$.Birthday\", \"$.LockoutEnabled\", \"$.AccessFailedCount\", \"$.LockoutEnd\", \"$.RowVersion\", \"$.Created\", \"$.Modified\", \"$.Deleted\"");
|
|
|
|
|
httpClient.DefaultRequestHeaders.Add("label", DateTime.Now.ToString());
|
|
|
|
|
var url = $"{_config.GetValue("doris","http://localhost:8030")}/api/example/User/_stream_load";
|
|
|
|
|
var server = _config.GetSection("doris").GetValue("server", "http://localhost:8030");
|
|
|
|
|
var database = _config.GetSection("doris").GetValue("database", "example");
|
|
|
|
|
var url = $"{server}/api/{database}/User/_stream_load";
|
|
|
|
|
|
|
|
|
|
using (var multiContent = new MultipartFormDataContent())
|
|
|
|
|
{
|
|
|
|
|
using (var ms = new MemoryStream())
|
|
|
|
|
{
|
|
|
|
|
using(var sw = new StreamWriter(ms))
|
|
|
|
|
using (var sw = new StreamWriter(ms))
|
|
|
|
|
{
|
|
|
|
|
sw.WriteLine("");
|
|
|
|
|
//{ "Id":"49fa11e7-4404-4fe5-9873-e89a01d8e529","UserName":"super1","SecurityStamp":"123456","PasswordHash":"579f889441b4a55d667233941d72a83ed644f7e5","PasswordConfirmed":1,"Email":"super@test.com","EmailConfirmed":1,"PhoneNumber":null,"PhoneNumberConfirmed":0,"RealName":null,"IdentityNumber":null,"IdentityConfirmed":0,"NickName":"超级管理员","Avatar":null,"Sex":null,"Birthday":null,"LockoutEnabled":0,"AccessFailedCount":0,"LockoutEnd":null,"RowVersion":"1e59e461-af12-446f-b876-eac8f58d3c79","Created":1618377036611675,"Modified":null,"Deleted":null}
|
|
|
|
|
sw.WriteLine("[");
|
|
|
|
|
for (int i = 0; i < list.Count; i++)
|
|
|
|
|
{
|
|
|
|
|
var item = list[i];
|
|
|
|
|
sw.Write($"{item}");
|
|
|
|
|
if (i < list.Count - 1)
|
|
|
|
|
{
|
|
|
|
|
sw.WriteLine(',');
|
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
{
|
|
|
|
|
sw.WriteLine();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
sw.WriteLine("]");
|
|
|
|
|
sw.Flush();
|
|
|
|
|
var data = ms.ToArray();
|
|
|
|
|
Debug.WriteLine(Encoding.UTF8.GetString(data));
|
|
|
|
|
var fileContent = new ByteArrayContent(data);
|
|
|
|
|
multiContent.Add(fileContent, "file", "input.csv");
|
|
|
|
|
var result = httpClient.PostAsync(url, multiContent).Result;//401 http basic
|
|
|
|
|
multiContent.Add(fileContent, "file", "example.json");
|
|
|
|
|
var result = httpClient.PutAsync(url, multiContent).Result;
|
|
|
|
|
if (result.StatusCode == System.Net.HttpStatusCode.OK)
|
|
|
|
|
{
|
|
|
|
|
var responseText = result.Content.ReadAsStringAsync().Result;
|
|
|
|
|
Debug.WriteLine(responseText);
|
|
|
|
|
var response = JsonDocument.Parse(responseText);
|
|
|
|
|
var label = response.RootElement.GetProperty("Label").GetRawText();
|
|
|
|
|
var message = response.RootElement.GetProperty("Message").GetRawText();
|
|
|
|
|
this._logger.LogInformation($"{label}:{message}");
|
|
|
|
|
var status = response.RootElement.GetProperty("Status").GetRawText();
|
|
|
|
|
if (status == "Success" || status == "Publish Timeout")
|
|
|
|
|
{
|
|
|
|
|
consumer.Commit();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
//{
|
|
|
|
|
// "TxnId": 3009,
|
|
|
|
|
// "Label": "c1f15903-805d-4d8a-8533-4c96080c49fe",
|
|
|
|
|
// "Status": "Success",
|
|
|
|
|
// "Message": "OK",
|
|
|
|
|
// "NumberTotalRows": 3,
|
|
|
|
|
// "NumberLoadedRows": 3,
|
|
|
|
|
// "NumberFilteredRows": 0,
|
|
|
|
|
// "NumberUnselectedRows": 0,
|
|
|
|
|
// "LoadBytes": 1678,
|
|
|
|
|
// "LoadTimeMs": 689,
|
|
|
|
|
// "BeginTxnTimeMs": 1,
|
|
|
|
|
// "StreamLoadPutTimeMs": 4,
|
|
|
|
|
// "ReadDataTimeMs": 0,
|
|
|
|
|
// "WriteDataTimeMs": 257,
|
|
|
|
|
// "CommitAndPublishTimeMs": 421
|
|
|
|
|
//}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
//httpClient.PutAsync
|
|
|
|
|
//consumer.Commit();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
catch (OperationCanceledException)
|
|
|
|
|
@ -129,7 +181,7 @@ namespace Kafka2Doris
|
|
|
|
|
this._logger.LogError(ex.ToString());
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
//
|
|
|
|
|
await Task.Delay(this._config.GetValue("delay", 1000 * 60), stoppingToken);
|
|
|
|
|
}
|
|
|
|
|
|
