Flink把kafa数据写入Doris的N种方法及对比。

温锦文欧普厨电及净水器总代理  金牌会员 | 2025-1-22 05:20:39 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 971|帖子 971|积分 2913

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要 登录 才可以下载或查看，没有账号？立即注册

x
用Flink+Doris来开发实时数仓，首要办理是怎样接入kafka实时流，下面是参考Doris官方文档和代码，在本身项目开发的实践中总结，包罗一些容易踩坑的细节。

Routine Load方法

如果Doris是2.1以上，不必要复杂的数据转换的，发起利用Doris自带的Routine Load，实测利用方便，性能高。
 接入kafka实时数据

Doris 可以通过 Routine Load 导入方式持续消费 Kafka Topic 中的数据。在提交 Routine Load 作业后，Doris 会持续运行该导入作业，实时生成导入任务不断消费 Kakfa 集群中指定 Topic 中的消息。
Routine Load 是一个流式导入作业，支持 Exactly-Once 语义，保证数据不丢不重。下面示例怎样通过拉入kafka数据(json格式）：
首先是创建必要导入的表：
  1. CREATE TABLE testdb.test_routineload_tbl(
  2.     user_id            BIGINT       NOT NULL COMMENT "user id",
  3.     name               VARCHAR(20)           COMMENT "name",
  4.     age                INT                   COMMENT "age"
  5. )
  6. DUPLICATE KEY(user_id)
  7. DISTRIBUTED BY HASH(user_id) BUCKETS 10;
复制代码
如果要接入的数据是主键不重复的，可以设置为Unique模型，如许可以删除或修改。

创建 Routine Load 导入作业
在 Doris 中，利用 CREATE ROUTINE LOAD 命令，创建导入作业
  1. CREATE ROUTINE LOAD testdb.example_routine_load_json ON test_routineload_tbl
  2. COLUMNS(user_id,name,age)
  3. PROPERTIES(
  4.     "format"="json",
  5.     "max_error_number" = "999999999999",
  6.      "strip_outer_array"="true",
  7.     "jsonpaths"="["$.user_id","$.name","$.age"]"
  8. )
  9. FROM KAFKA(
  10.     "kafka_broker_list" = "192.168.88.62:9092",
  11.     "kafka_topic" = "test-routine-load-json",
  12.     "property.kafka_default_offsets" = "OFFSET_BEGINNING"
  13. );
复制代码
踩坑的问题细节 

max_error_number采样窗口内，答应的最大错误行数。必须大于便是 0。默认是 0，即不答应有错误行。采样窗口为 max_batch_rows * 10。即如果在采样窗口内，错误行数大于 max_error_number，则会导致例行作业被停息，必要人工介入检查数据质量问题，通过 SHOW ROUTINE LOAD 命令中 ErrorLogUrls 检查数据的质量问题。被 where 条件过滤掉的行不算错误行。
strip_outer_array当导入数据格式为 json 时，strip_outer_array 为 true 表示 JSON 数据以数组的情势展现，数据中的每一个元素将被视为一行数据。默认值是 false。通常情况下，Kafka 中的 JSON 数据可能以数组情势表示，即在最外层中包罗中括号[]，此时，可以指定 "strip_outer_array" = "true"，以数组模式消费 Topic 中的数据。如以下数据会被剖析成两行：[{"user_id":1,"name":"Emily","age":25},{"user_id":2,"name":"Benjamin","age":35}]
  1. 我公司的kafka，是需要设置"strip_outer_array"="true"，根据实际来调整。
  3. 有些kafka的数据有脏数据，可以用max_error_number来过滤。或者考虑用脚本来检测：
复制代码
  1. import pymysql  #导入 pymysql
  2. import requests,json
  5. #打开数据库连接
  6. db= pymysql.connect(host="host",user="user",
  7.                     password="passwd",db="database",port=port)
  9. # 使用cursor()方法获取操作游标
  10. cur = db.cursor()
  12. #1.查询操作
  13. # 编写sql 查询语句
  14. sql = "show routine load"
  15. cur.execute(sql)        #执行sql语句
  16. results = cur.fetchall()        #获取查询的所有记录
  17. for row in results :
  18.     name = row[1]
  19.     state = row[7]
  20.     if state != 'RUNNING':
  21.         err_log_urls = row[16]
  22.         reason_state_changed = row[15]
  23.         msg = "doris 数据导入任务异常:\n name=%s \n state=%s \n reason_state_changed=%s \n err_log_urls=%s \n即将自动恢复，请检查错误信息" % (name, state,
  24. reason_state_changed, err_log_urls)
  25.         payload_message = {
  26.     "msg_type": "text",
  27.     "content": {
  28.         "text": msg
  29.     }
  30. }
  31.         url = 'lark 报警url'
  32.         s = json.dumps(payload_message)
  33.         r = requests.post(url, data=s)
  34.         cur.execute("resume routine load for " + name)
  36. cur.close()
  37. db.close()
复制代码
  1. 通过 SHOW ROUTINE LOAD 来查看结果
复制代码
  1. mysql> SHOW ROUTINE LOAD FOR testdb.example_routine_load
复制代码
  1. 我是没按官方文档加\G，加了\G报错。
  3. 如果导入数据有问题，可以通过上面的命令，查看下面这2个：
复制代码
ErrorLogUrls被过滤的质量不合格的数据的检察所在
OtherMsg其他错误信息
  1. 具体可以参考：Routine Load - Apache Doris
复制代码

Flink Doris Connector方法

Flink Doris Connector 可以支持通过 Flink 操纵（读取、插入、修改、删除）Doris 中存储的数据。
这种方法适合Doris2.1以下的，大概必要对kafka数据举行复杂过滤或数据转换。Routine Load方法也是可以简单过滤的，但觉得还是不可以或许高度定制：
  1. CREATE ROUTINE LOAD demo.kafka_job04 ON routine_test04
  2.         COLUMNS TERMINATED BY ",",
  3.         WHERE id >= 3
  4.         FROM KAFKA
  5.         (
  6.             "kafka_broker_list" = "10.16.10.6:9092",
  7.             "kafka_topic" = "routineLoad04",
  8.             "property.kafka_default_offsets" = "OFFSET_BEGINNING"
  9.         );  
复制代码
由于公司项目flink是很老版本1.11，如果是新版本flink，请参考官方文档举行调整：Flink Doris Connector - Apache Doris
完整示例

pom依赖如下：
  1. <properties>
  2.         <scala.version>2.12.10</scala.version>
  3.         <scala.binary.version>2.12</scala.binary.version>
  4.         <java.version>1.8</java.version>
  5.         <flink.version>1.12.2</flink.version>
  6.         <fastjson.version>1.2.83</fastjson.version>
  7.         <hadoop.version>2.8.3</hadoop.version>
  8.         <scope.mode>compile</scope.mode>
  9.     </properties>
  10.     <dependencies>
  11.         <dependency>
  12.             <groupId>org.apache.flink</groupId>
  13.             <artifactId>flink-table-api-scala-bridge_${scala.binary.version}</artifactId>
  14.             <version>${flink.version}</version>
  15.         </dependency>
  16.         <dependency>
  17.             <groupId>org.apache.flink</groupId>
  18.             <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
  19.             <version>${flink.version}</version>
  20.         </dependency>
  21.         <dependency>
  22.             <groupId>org.apache.flink</groupId>
  23.             <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
  24.             <version>${flink.version}</version>
  25.         </dependency>
  26.         <dependency>
  27.             <groupId>mysql</groupId>
  28.             <artifactId>mysql-connector-java</artifactId>
  29.             <version>8.0.12</version>
  30.         </dependency>
  31.         <dependency>
  32.             <groupId>com.alibaba.ververica</groupId>
  33.             <artifactId>flink-connector-mysql-cdc</artifactId>
  34.             <version>1.3.0</version>
  35.         </dependency>
  36.         <dependency>
  37.             <groupId>org.apache.flink</groupId>
  38.             <artifactId>flink-clients_2.11</artifactId>
  39.             <version>${flink.version}</version>
  40.         </dependency>
  41.         <dependency>
  42.             <groupId>org.apache.flink</groupId>
  43.             <artifactId>flink-connector-jdbc_2.11</artifactId>
  44.             <version>${flink.version}</version>
  45.         </dependency>
  46.         <dependency>
  47.             <groupId>org.apache.flink</groupId>
  48.             <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
  49.             <version>${flink.version}</version>
  50.         </dependency>
  51.         <dependency>
  52.             <groupId>com.alibaba</groupId>
  53.             <artifactId>fastjson</artifactId>
  54.             <version>${fastjson.version}</version>
  55.         </dependency>
  56.         <dependency>
  57.             <groupId>org.apache.flink</groupId>
  58.             <artifactId>flink-hadoop-compatibility_2.11</artifactId>
  59.             <version>${flink.version}</version>
  60.         </dependency>
  61.         <dependency>
  62.             <groupId>org.apache.hadoop</groupId>
  63.             <artifactId>hadoop-client</artifactId>
  64.             <version>${hadoop.version}</version>
  65.         </dependency>
  66.         <dependency>
  67.             <groupId>org.apache.flink</groupId>
  68.             <artifactId>flink-shaded-hadoop-2-uber</artifactId>
  69.             <version>2.6.5-10.0</version>
  70.         </dependency>
  71.         <dependency>
  72.             <groupId>org.apache.doris</groupId>
  73.             <artifactId>flink-doris-connector-1.14_2.12</artifactId>
  74.             <version>1.0.3</version>
  75.         </dependency>
  76.         <!--After adding the following two dependencies, Flink's log will appear-->
  77.         <dependency>
  78.             <groupId>org.slf4j</groupId>
  79.             <artifactId>slf4j-api</artifactId>
  80.             <version>1.7.25</version>
  81.         </dependency>
  82.         <dependency>
  83.             <groupId>org.slf4j</groupId>
  84.             <artifactId>slf4j-simple</artifactId>
  85.             <version>1.7.25</version>
  86.         </dependency>
  87.       
  88.         <dependency>
  89.             <groupId>org.apache.httpcomponents</groupId>
  90.             <artifactId>httpclient</artifactId>
  91.             <version>4.5.13</version>
  92.         </dependency>
  93.         <dependency>
  94.             <groupId>org.apache.httpcomponents</groupId>
  95.             <artifactId>httpcore</artifactId>
  96.             <version>4.4.12</version>
  97.         </dependency>
  98.     </dependencies>
复制代码

 FlinkKafka2Doris ：
  1. public class FlinkKafka2Doris {
  2.     //kafka address
  3.     private static final String bootstrapServer = "xxx:9092,xxx:9092,xxx:9092";
  4.     //kafka groupName
  5.     private static final String groupName = "test_flink_doris_group";
  6.     //kafka topicName
  7.     private static final String topicName = "test_flink_doris";
  8.     //doris ip port
  9.     private static final String hostPort = "xxx:8030";
  10.     //doris dbName
  11.     private static final String dbName = "test1";
  12.     //doris tbName
  13.     private static final String tbName = "doris_test_source_2";
  14.     //doris userName
  15.     private static final String userName = "root";
  16.     //doris password
  17.     private static final String password = "";
  18.     //doris columns
  19.     private static final String columns = "name,age,price,sale";
  20.     //json format
  21.     private static final String jsonFormat = "["$.name","$.age","$.price","$.sale"]";
  23.     public static void main(String[] args) throws Exception {
  25.         Properties props = new Properties();
  26.         props.put("bootstrap.servers", bootstrapServer);
  27.         props.put("group.id", groupName);
  28.         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  29.         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  30.         props.put("auto.offset.reset", "earliest");
  31.         props.put("max.poll.records", "10000");
  33.         StreamExecutionEnvironment blinkStreamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
  34.         blinkStreamEnv.enableCheckpointing(10000);
  35.         blinkStreamEnv.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  37.         FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<>(topicName,
  38.                 new SimpleStringSchema(),
  39.                 props);
  41.         DataStreamSource<String> dataStreamSource = blinkStreamEnv.addSource(flinkKafkaConsumer);
  43. //在这里进行数据过滤或转换
  45.         DorisStreamLoad dorisStreamLoad = new DorisStreamLoad(hostPort, dbName, tbName, userName, password);
  47.         dataStreamSource.addSink(new DorisSink(dorisStreamLoad,columns,jsonFormat));
  49.         blinkStreamEnv.execute("flink kafka to doris");
  51.     }
  52. }
复制代码

DorisStreamLoad：
  1. import org.slf4j.Logger;
  2. import org.slf4j.LoggerFactory;
  5. import java.io.Serializable;
  6. import java.io.IOException;
  7. import java.io.BufferedOutputStream;
  8. import java.io.InputStream;
  9. import java.io.BufferedReader;
  10. import java.io.InputStreamReader;
  11. import java.net.HttpURLConnection;
  12. import java.net.URL;
  13. import java.nio.charset.StandardCharsets;
  14. import java.util.Base64;
  15. import java.util.Calendar;
  16. import java.util.UUID;
  19. /**
  20. * doris streamLoad
  21. */
  23. public class DorisStreamLoad implements Serializable {
  25.     private static final Logger log = LoggerFactory.getLogger(DorisStreamLoad.class);
  27.     private static String loadUrlPattern = "http://%s/api/%s/%s/_stream_load?";
  28.     private String hostPort;
  29.     private String db;
  30.     private String tbl;
  31.     private String user;
  32.     private String passwd;
  33.     private String loadUrlStr;
  34.     private String authEncoding;
  37.     public DorisStreamLoad(String hostPort, String db, String tbl, String user, String passwd) {
  38.         this.hostPort = hostPort;
  39.         this.db = db;
  40.         this.tbl = tbl;
  41.         this.user = user;
  42.         this.passwd = passwd;
  43.         this.loadUrlStr = String.format(loadUrlPattern, hostPort, db, tbl);
  44.         this.authEncoding = Base64.getEncoder().encodeToString(String.format("%s:%s", user, passwd).getBytes(StandardCharsets.UTF_8));
  45.     }
  47.     private HttpURLConnection getConnection(String urlStr, String label, String columns, String jsonformat) throws IOException {
  48.         URL url = new URL(urlStr);
  49.         HttpURLConnection conn = (HttpURLConnection) url.openConnection();
  50.         conn.setInstanceFollowRedirects(false);
  51.         conn.setRequestMethod("PUT");
  52.         conn.setRequestProperty("Authorization", "Basic " + authEncoding);
  53.         conn.addRequestProperty("Expect", "100-continue");
  54.         conn.addRequestProperty("Content-Type", "text/plain; charset=UTF-8");
  55.         conn.addRequestProperty("label", label);
  56.         conn.addRequestProperty("max_filter_ratio", "0");
  57.         conn.addRequestProperty("strict_mode", "true");
  58.         conn.addRequestProperty("columns", columns);
  59.         conn.addRequestProperty("format", "json");
  60.         conn.addRequestProperty("jsonpaths", jsonformat);
  61.         conn.addRequestProperty("strip_outer_array", "true");
  62.         conn.setDoOutput(true);
  63.         conn.setDoInput(true);
  65.         return conn;
  66.     }
  68.     public static class LoadResponse {
  69.         public int status;
  70.         public String respMsg;
  71.         public String respContent;
  73.         public LoadResponse(int status, String respMsg, String respContent) {
  74.             this.status = status;
  75.             this.respMsg = respMsg;
  76.             this.respContent = respContent;
  77.         }
  79.         @Override
  80.         public String toString() {
  81.             StringBuilder sb = new StringBuilder();
  82.             sb.append("status: ").append(status);
  83.             sb.append(", resp msg: ").append(respMsg);
  84.             sb.append(", resp content: ").append(respContent);
  85.             return sb.toString();
  86.         }
  87.     }
  89.     public LoadResponse loadBatch(String data, String columns, String jsonformat) {
  90.         Calendar calendar = Calendar.getInstance();
  91.         String label = String.format("audit_%s%02d%02d_%02d%02d%02d_%s",
  92.                 calendar.get(Calendar.YEAR), calendar.get(Calendar.MONTH) + 1, calendar.get(Calendar.DAY_OF_MONTH),
  93.                 calendar.get(Calendar.HOUR_OF_DAY), calendar.get(Calendar.MINUTE), calendar.get(Calendar.SECOND),
  94.                 UUID.randomUUID().toString().replaceAll("-", ""));
  96.         HttpURLConnection feConn = null;
  97.         HttpURLConnection beConn = null;
  98.         try {
  99.             // build request and send to fe
  100.             feConn = getConnection(loadUrlStr, label, columns, jsonformat);
  101.             int status = feConn.getResponseCode();
  102.             // fe send back http response code TEMPORARY_REDIRECT 307 and new be location
  103.             if (status != 307) {
  104.                 throw new Exception("status is not TEMPORARY_REDIRECT 307, status: " + status);
  105.             }
  106.             String location = feConn.getHeaderField("Location");
  107.             if (location == null) {
  108.                 throw new Exception("redirect location is null");
  109.             }
  110.             // build request and send to new be location
  111.             beConn = getConnection(location, label, columns, jsonformat);
  112.             // send data to be
  113.             BufferedOutputStream bos = new BufferedOutputStream(beConn.getOutputStream());
  114.             bos.write(data.getBytes());
  115.             bos.close();
  117.             // get respond
  118.             status = beConn.getResponseCode();
  119.             String respMsg = beConn.getResponseMessage();
  120.             InputStream stream = (InputStream) beConn.getContent();
  121.             BufferedReader br = new BufferedReader(new InputStreamReader(stream));
  122.             StringBuilder response = new StringBuilder();
  123.             String line;
  124.             while ((line = br.readLine()) != null) {
  125.                 response.append(line);
  126.             }
  127.             return new LoadResponse(status, respMsg, response.toString());
  129.         } catch (Exception e) {
  130.             e.printStackTrace();
  131.             String err = "failed to load audit via AuditLoader plugin with label: " + label;
  132.             log.warn(err, e);
  133.             return new LoadResponse(-1, e.getMessage(), err);
  134.         } finally {
  135.             if (feConn != null) {
  136.                 feConn.disconnect();
  137.             }
  138.             if (beConn != null) {
  139.                 beConn.disconnect();
  140.             }
  141.         }
  142.     }
  144. }
复制代码

DorisSink：
  1. import com.alibaba.fastjson.JSON;
  2. import org.apache.flink.configuration.Configuration;
  3. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
  4. import org.slf4j.Logger;
  5. import org.slf4j.LoggerFactory;
  7. import java.util.ArrayList;
  8. import java.util.Arrays;
  9. import java.util.List;
  11. /**
  12. * Custom doris sink
  13. */
  14. public class DorisSink extends RichSinkFunction<String> {
  17.     private static final Logger log = LoggerFactory.getLogger(DorisSink.class);
  19.     private final static List<String> DORIS_SUCCESS_STATUS = new ArrayList<>(Arrays.asList("Success", "Publish Timeout"));
  21.     private DorisStreamLoad dorisStreamLoad;
  23.     private String columns;
  25.     private String jsonFormat;
  27.     public DorisSink(DorisStreamLoad dorisStreamLoad, String columns, String jsonFormat) {
  28.         this.dorisStreamLoad = dorisStreamLoad;
  29.         this.columns = columns;
  30.         this.jsonFormat = jsonFormat;
  31.     }
  33.     @Override
  34.     public void open(Configuration parameters) throws Exception {
  35.         super.open(parameters);
  36.     }
  39.     /**
  40.      * Determine whether StreamLoad is successful
  41.      *
  42.      * @param respContent streamLoad returns the entity
  43.      * @return
  44.      */
  45.     public static Boolean checkStreamLoadStatus(RespContent respContent) {
  46.         if (DORIS_SUCCESS_STATUS.contains(respContent.getStatus())
  47.                 && respContent.getNumberTotalRows() == respContent.getNumberLoadedRows()) {
  48.             return true;
  49.         } else {
  50.             return false;
  51.         }
  52.     }
  54.     @Override
  55.     public void invoke(String value, Context context) throws Exception {
  56.         DorisStreamLoad.LoadResponse loadResponse = dorisStreamLoad.loadBatch(value, columns, jsonFormat);
  57.         if (loadResponse != null && loadResponse.status == 200) {
  58.             RespContent respContent = JSON.parseObject(loadResponse.respContent, RespContent.class);
  59.             if (!checkStreamLoadStatus(respContent)) {
  60.                 log.error("Stream Load fail{}:", loadResponse);
  61.             }
  62.         } else {
  63.             log.error("Stream Load Request failed:{}", loadResponse);
  64.         }
  65.     }
  66. }
复制代码

RespContent：
  1. import java.io.Serializable;
  3. /**
  4. * Entity returned by streamLoad
  5. */
  6. public class RespContent implements Serializable {
  8.     private static final long serialVersionUID = 1L;
  11.     /**
  12.      * Imported transaction ID. The user may not perceive it.
  13.      */
  14.     private int TxnId;
  15.     /**
  16.      * Import Label. Specified by the user or automatically generated by the system.
  17.      */
  18.     private String Label;
  19.     /**
  20.      * Import complete status.
  21.      * "Success": Indicates that the import was successful.
  22.      * "Publish Timeout": This status also indicates that the import has been completed, but the data may be visible with a delay, and there is no need to retry.
  23.      * "Label Already Exists": The Label is duplicated, and the Label needs to be replaced.
  24.      */
  25.     private String Status;
  26.     /**
  27.      * The status of the import job corresponding to the existing Label.
  28.      * This field will only be displayed when the Status is "Label Already Exists".
  29.      * The user can know the status of the import job corresponding to the existing Label through this status.
  30.      * "RUNNING" means that the job is still executing, and "FINISHED" means that the job is successful.
  31.      */
  32.     private String ExistingJobStatus;
  33.     /**
  34.      * Import error information
  35.      */
  36.     private String Message;
  37.     /**
  38.      * Import the total number of processed rows
  39.      */
  40.     private long NumberTotalRows;
  41.     /**
  42.      * The number of rows successfully imported.
  43.      */
  44.     private long NumberLoadedRows;
  45.     /**
  46.      * Number of rows with unqualified data quality。
  47.      */
  48.     private int NumberFilteredRows;
  49.     /**
  50.      * The number of rows filtered by the where condition
  51.      */
  52.     private int NumberUnselectedRows;
  53.     /**
  54.      * Number of bytes imported。
  55.      */
  56.     private long LoadBytes;
  57.     /**
  58.      * Import completion time. The unit is milliseconds.
  59.      */
  60.     private int LoadTimeMs;
  61.     /**
  62.      * The time it takes to request Fe to start a transaction, in milliseconds
  63.      */
  64.     private int BeginTxnTimeMs;
  65.     /**
  66.      * The time it takes to request Fe to obtain the import data execution plan, in milliseconds
  67.      */
  68.     private int StreamLoadPutTimeMs;
  69.     /**
  70.      * The time spent reading data, in milliseconds
  71.      */
  72.     private int ReadDataTimeMs;
  73.     /**
  74.      * Time to perform a data write operation takes milliseconds。
  75.      */
  76.     private int WriteDataTimeMs;
  77.     /**
  78.      * The time taken to submit and publish the transaction request to Fe, in milliseconds
  79.      */
  80.     private int CommitAndPublishTimeMs;
  81.     /**
  82.      * If there is a data quality problem, check the specific error line by visiting this URL
  83.      */
  84.     private String ErrorURL;
  86.     public int getTxnId() {
  87.         return TxnId;
  88.     }
  90.     public void setTxnId(int txnId) {
  91.         TxnId = txnId;
  92.     }
  94.     public String getLabel() {
  95.         return Label;
  96.     }
  98.     public void setLabel(String label) {
  99.         Label = label;
  100.     }
  102.     public String getStatus() {
  103.         return Status;
  104.     }
  106.     public void setStatus(String status) {
  107.         Status = status;
  108.     }
  110.     public String getExistingJobStatus() {
  111.         return ExistingJobStatus;
  112.     }
  114.     public void setExistingJobStatus(String existingJobStatus) {
  115.         ExistingJobStatus = existingJobStatus;
  116.     }
  118.     public String getMessage() {
  119.         return Message;
  120.     }
  122.     public void setMessage(String message) {
  123.         Message = message;
  124.     }
  126.     public long getNumberTotalRows() {
  127.         return NumberTotalRows;
  128.     }
  130.     public void setNumberTotalRows(long numberTotalRows) {
  131.         NumberTotalRows = numberTotalRows;
  132.     }
  134.     public long getNumberLoadedRows() {
  135.         return NumberLoadedRows;
  136.     }
  138.     public void setNumberLoadedRows(long numberLoadedRows) {
  139.         NumberLoadedRows = numberLoadedRows;
  140.     }
  142.     public int getNumberFilteredRows() {
  143.         return NumberFilteredRows;
  144.     }
  146.     public void setNumberFilteredRows(int numberFilteredRows) {
  147.         NumberFilteredRows = numberFilteredRows;
  148.     }
  150.     public int getNumberUnselectedRows() {
  151.         return NumberUnselectedRows;
  152.     }
  154.     public void setNumberUnselectedRows(int numberUnselectedRows) {
  155.         NumberUnselectedRows = numberUnselectedRows;
  156.     }
  158.     public long getLoadBytes() {
  159.         return LoadBytes;
  160.     }
  162.     public void setLoadBytes(long loadBytes) {
  163.         LoadBytes = loadBytes;
  164.     }
  166.     public int getLoadTimeMs() {
  167.         return LoadTimeMs;
  168.     }
  170.     public void setLoadTimeMs(int loadTimeMs) {
  171.         LoadTimeMs = loadTimeMs;
  172.     }
  174.     public int getBeginTxnTimeMs() {
  175.         return BeginTxnTimeMs;
  176.     }
  178.     public void setBeginTxnTimeMs(int beginTxnTimeMs) {
  179.         BeginTxnTimeMs = beginTxnTimeMs;
  180.     }
  182.     public int getStreamLoadPutTimeMs() {
  183.         return StreamLoadPutTimeMs;
  184.     }
  186.     public void setStreamLoadPutTimeMs(int streamLoadPutTimeMs) {
  187.         StreamLoadPutTimeMs = streamLoadPutTimeMs;
  188.     }
  190.     public int getReadDataTimeMs() {
  191.         return ReadDataTimeMs;
  192.     }
  194.     public void setReadDataTimeMs(int readDataTimeMs) {
  195.         ReadDataTimeMs = readDataTimeMs;
  196.     }
  198.     public int getWriteDataTimeMs() {
  199.         return WriteDataTimeMs;
  200.     }
  202.     public void setWriteDataTimeMs(int writeDataTimeMs) {
  203.         WriteDataTimeMs = writeDataTimeMs;
  204.     }
  206.     public int getCommitAndPublishTimeMs() {
  207.         return CommitAndPublishTimeMs;
  208.     }
  210.     public void setCommitAndPublishTimeMs(int commitAndPublishTimeMs) {
  211.         CommitAndPublishTimeMs = commitAndPublishTimeMs;
  212.     }
  214.     public String getErrorURL() {
  215.         return ErrorURL;
  216.     }
  218.     public void setErrorURL(String errorURL) {
  219.         ErrorURL = errorURL;
  220.     }
  222.     @Override
  223.     public String toString() {
  224.         return "RespContent{" +
  225.                 "TxnId=" + TxnId +
  226.                 ", Label='" + Label + '\'' +
  227.                 ", Status='" + Status + '\'' +
  228.                 ", ExistingJobStatus='" + ExistingJobStatus + '\'' +
  229.                 ", Message='" + Message + '\'' +
  230.                 ", NumberTotalRows=" + NumberTotalRows +
  231.                 ", NumberLoadedRows=" + NumberLoadedRows +
  232.                 ", NumberFilteredRows=" + NumberFilteredRows +
  233.                 ", NumberUnselectedRows=" + NumberUnselectedRows +
  234.                 ", LoadBytes=" + LoadBytes +
  235.                 ", LoadTimeMs=" + LoadTimeMs +
  236.                 ", BeginTxnTimeMs=" + BeginTxnTimeMs +
  237.                 ", StreamLoadPutTimeMs=" + StreamLoadPutTimeMs +
  238.                 ", ReadDataTimeMs=" + ReadDataTimeMs +
  239.                 ", WriteDataTimeMs=" + WriteDataTimeMs +
  240.                 ", CommitAndPublishTimeMs=" + CommitAndPublishTimeMs +
  241.                 ", ErrorURL='" + ErrorURL + '\'' +
  242.                 '}';
  243.     }
  244. }
复制代码


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

温锦文欧普厨电及净水器总代理

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

AI 运维 CIO 存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表