C# .Net Core通过StreamLoad向Doris写入CSV数据

以下代码可以只关注StreamLoad具体实现。

1.创建StreamLoad返回值Model

public class StreamLoadResponse
{public long TxnId { get; set; }public string Label { get; set; }public string Comment { get; set; }public string TwoPhaseCommit { get; set; }public string Status { get; set; }public string Message { get; set; }public long NumberTotalRows { get; set; }public long NumberLoadedRows { get; set; }public long NumberFilteredRows { get; set; }public long NumberUnselectedRows { get; set; }public long LoadBytes { get; set; }public long LoadTimeMs { get; set; }public long BeginTxnTimeMs { get; set; }public long StreamLoadPutTimeMs { get; set; }public long ReadDataTimeMs { get; set; }public long WriteDataTimeMs { get; set; }public long CommitAndPublishTimeMs { get; set; }
}

2.创建Doris StreamLoad接口

public interface IDorisApiService
{/// <summary>/// /// </summary>/// <param name="database">数据库</param>/// <param name="table">表</param>/// <param name="authorization">认证信息，格式 username:pwd</param>/// <param name="content">csv格式的字符串</param>/// <returns></returns>StreamLoadResponse StreamLoad(string database, string table, string authorization, string content);
}

3.实现接口，核心代码，逻辑并不复杂，组装一个http请求所需的内容。

需要注意的是：（1）示例csv格式的字符串分割符为‘\t’，而不是常用的逗号，这也是官方默认的分割方式，如果你想用其他的分隔符，需要在header里配置column_separator。建议不要用逗号，因为涉及到复杂的json字符串的时候，里面的逗号会导致解析异常，即便官方文档里有相关的处理方式（enclose）,似乎仍然存在问题。（2）我们请求了两次，第一次请求会重定向到BE节点的地址，然后用此地址再次请求。这是正常的。(3)我们采用的format是csv_with_names，第一行是列明，请确保跟数据库table列顺序和数量保持一致

public class DorisApiService : IDorisApiService
{private readonly HttpClient _httpClient;public DorisApiService(HttpClient httpClient){_httpClient = httpClient;}public StreamLoadResponse StreamLoad(string database, string table, string authorization, string content){var url = $"/api/{database}/{table}/_stream_load";var request = new HttpRequestMessage(HttpMethod.Put, url);request.Headers.Add("Expect", "100-continue");request.Headers.Add("format", "csv_with_names");request.Headers.Add("column_separator", "\t");request.Headers.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Basic", authorization);var response = _httpClient.Send(request, HttpCompletionOption.ResponseHeadersRead);if (response.StatusCode == HttpStatusCode.TemporaryRedirect || response.StatusCode == HttpStatusCode.RedirectKeepVerb){var redirectUrl = response.Headers.Location.ToString();request = new HttpRequestMessage(HttpMethod.Put, redirectUrl){Content = new StringContent(content, Encoding.UTF8, "text/plain")};request.Headers.Add("Expect", "100-continue");request.Headers.Add("format", "csv_with_names");request.Headers.Add("column_separator", "\t");request.Headers.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Basic", authorization);response = _httpClient.Send(request, HttpCompletionOption.ResponseHeadersRead);}string responseBody = response.Content.ReadAsStringAsync().Result;           if (response.IsSuccessStatusCode){var streamLoadResponse = JsonSerializer.Deserialize<StreamLoadResponse>(responseBody);if (streamLoadResponse.Status == "Success"){return streamLoadResponse;}else{throw new Exception(responseBody);}}else{throw new Exception(responseBody);}}
}

4.Program配置

services.AddHttpClient<IDorisApiService, DorisApiService>(client =>
{//从配置文件获取Doris的请求地址和端口：settings.ApiHostclient.BaseAddress = new Uri(settings.ApiHost);client.Timeout = TimeSpan.FromSeconds(300);
}).ConfigurePrimaryHttpMessageHandler(() => new HttpClientHandler
{AllowAutoRedirect = false
});