DTStack / DTStack/chunjun

flinkx1.12性能问题:mysql-hive 数据量9000万+

Open
#1,136 8 comments 0 reactions 0 assignees View on GitHub
bug
Dominant language
Java
Stars
4.1k
Forks
1.7k
PR merge metrics
No merged PRs in 30d

Description

### Search before asking

- [X] I had searched in the [issues](https://github.com/DTStack/chunjun/issues) and found no similar issues.

### What happened

问题描述:
1、来源数据为700万+ 时,通过跳针flink job相关参数,可以在6分钟之内处理完
2、来源为9000万+时,mysql-hive同步时,来源端mysqlsource无法拉取数据,是因为source端需要一次查询数据 无法范围rs
来源端执行SQL:
SELECT `id`, `device_id`, `point`, `hash`, `value`, `acq_time`, `ratio`, `max_val`, `min_val`, `out_param`, `item_code`, `descr`, `save_hst`, `del_flag`, `create_user`, `update_user`, `create_time`, `update_time`, `ext_first`, `ext_second`, `ext_third`, `ext_fourth`, `ext_fifth`, `ext_sixth`, `ext_seventh`, `ext_eighth`, `ext_ninth`, `ext_tenth`, `device_add_id`, `today_start_value`, `unit` FROM `tb_device_point_data_hst_01` WHERE 1=1

====================================================================================
Caused by: java.lang.IllegalArgumentException: open() failed.The last packet successfully received from the server was 373,982 milliseconds ago. The last packet sent successfully to the server was 373,983 milliseconds ago. is longer than the server configured value of 'wait_timeout'. You should consider either expiring and/or testing connection validity before use in your application, increasing the server configured values for client timeouts, or using the Connector/J connection property 'autoReconnect=true' to avoid this problem.
querySQL: SELECT `id`, `device_id`, `point`, `hash`, `value`, `acq_time`, `ratio`, `max_val`, `min_val`, `out_param`, `item_code`, `descr`, `save_hst`, `del_flag`, `create_user`, `update_user`, `create_time`, `update_time`, `ext_first`, `ext_second`, `ext_third`, `ext_fourth`, `ext_fifth`, `ext_sixth`, `ext_seventh`, `ext_eighth`, `ext_ninth`, `ext_tenth`, `device_add_id`, `today_start_value`, `unit` FROM `tb_device_point_data_hst_01` WHERE 1=1
at com.dtstack.flinkx.connector.jdbc.source.JdbcInputFormat.openInternal(JdbcInputFormat.java:139)
at com.dtstack.flinkx.source.format.BaseRichInputFormat.open(BaseRichInputFormat.java:148)
at com.dtstack.flinkx.source.DtInputFormatSourceFunction.run(DtInputFormatSourceFunction.java:126)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:110)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:66)
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:263)
Caused by: com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: The last packet successfully received from the server was 373,982 milliseconds ago. The last packet sent successfully to the server was 373,983 milliseconds ago. is longer than the server configured value of 'wait_timeout'. You should consider either expiring and/or testing connection validity before use in your application, increasing the server configured values for client timeouts, or using the Connector/J connection property 'autoReconnect=true' to avoid this problem.
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
at com.mysql.jdbc.Util.handleNewInstance(Util.java:425)
at com.mysql.jdbc.SQLError.createCommunicationsException(SQLError.java:990)
at com.mysql.jdbc.MysqlIO.nextRowFast(MysqlIO.java:2232)
at com.mysql.jdbc.MysqlIO.nextRow(MysqlIO.java:1992)
at com.mysql.jdbc.MysqlIO.readSingleRowSet(MysqlIO.java:3413)
at com.mysql.jdbc.MysqlIO.getResultSet(MysqlIO.java:471)
at com.mysql.jdbc.MysqlIO.readResultsForQueryOrUpdate(MysqlIO.java:3115)
at com.mysql.jdbc.MysqlIO.readAllResults(MysqlIO.java:2344)
at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2739)
at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2482)
at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2440)
at com.mysql.jdbc.StatementImpl.executeQuery(StatementImpl.java:1381)
at com.dtstack.flinkx.connector.jdbc.source.JdbcInputFormat.executeQuery(JdbcInputFormat.java:769)
at com.dtstack.flinkx.connector.jdbc.source.JdbcInputFormat.openInternal(JdbcInputFormat.java:122)
... 5 common frames omitted
Caused by: java.io.EOFException: Can not read response from server. Expected to read 255 bytes, read 0 bytes before connection was unexpectedly lost.
at com.mysql.jdbc.MysqlIO.readFully(MysqlIO.java:3014)
at com.mysql.jdbc.MysqlIO.nextRowFast(MysqlIO.java:2215)
... 16 common frames omitted

### What you expected to happen

{
"job": {
"content": [{
"reader": {
"parameter": {
"password": "root",
"dataSourceId": 14,
"column": [{
"precision": 20,
"name": "id",
"columnDisplaySize": 20,
"type": "BIGINT"
}, {
"precision": 32,
"name": "device_id",
"columnDisplaySize": 32,
"type": "VARCHAR"
}, {
"precision": 12,
"name": "point",
"columnDisplaySize": 12,
"type": "VARCHAR"
}, {
"precision": 32,
"name": "hash",
"columnDisplaySize": 32,
"type": "VARCHAR"
}, {
"precision": 32,
"name": "value",
"columnDisplaySize": 32,
"type": "VARCHAR"
}, {
"precision": 19,
"name": "acq_time",
"columnDisplaySize": 19,
"type": "DATETIME"
}, {
"precision": 12,
"name": "ratio",
"columnDisplaySize": 12,
"type": "FLOAT"
}, {
"precision": 32,
"name": "max_val",
"columnDisplaySize": 32,
"type": "VARCHAR"
}, {
"precision": 32,
"name": "min_val",
"columnDisplaySize": 32,
"type": "VARCHAR"
}, {
"precision": 32,
"name": "out_param",
"columnDisplaySize": 32,
"type": "VARCHAR"
}, {
"precision": 8,
"name": "item_code",
"columnDisplaySize": 8,
"type": "VARCHAR"
}, {
"precision": 32,
"name": "descr",
"columnDisplaySize": 32,
"type": "VARCHAR"
}, {
"precision": 16,
"name": "save_hst",
"columnDisplaySize": 16,
"type": "SMALLINT"
}, {
"precision": 1,
"name": "del_flag",
"columnDisplaySize": 1,
"type": "TINYINT"
}, {
"precision": 20,
"name": "create_user",
"columnDisplaySize": 20,
"type": "BIGINT"
}, {
"precision": 20,
"name": "update_user",
"columnDisplaySize": 20,
"type": "BIGINT"
}, {
"precision": 19,
"name": "create_time",
"columnDisplaySize": 19,
"type": "DATETIME"
}, {
"precision": 19,
"name": "update_time",
"columnDisplaySize": 19,
"type": "DATETIME"
}, {
"precision": 50,
"name": "ext_first",
"columnDisplaySize": 50,
"type": "VARCHAR"
}, {
"precision": 50,
"name": "ext_second",
"columnDisplaySize": 50,
"type": "VARCHAR"
}, {
"precision": 50,
"name": "ext_third",
"columnDisplaySize": 50,
"type": "VARCHAR"
}, {
"precision": 50,
"name": "ext_fourth",
"columnDisplaySize": 50,
"type": "VARCHAR"
}, {
"precision": 50,
"name": "ext_fifth",
"columnDisplaySize": 50,
"type": "VARCHAR"
}, {
"precision": 50,
"name": "ext_sixth",
"columnDisplaySize": 50,
"type": "VARCHAR"
}, {
"precision": 50,
"name": "ext_seventh",
"columnDisplaySize": 50,
"type": "VARCHAR"
}, {
"precision": 50,
"name": "ext_eighth",
"columnDisplaySize": 50,
"type": "VARCHAR"
}, {
"precision": 50,
"name": "ext_ninth",
"columnDisplaySize": 50,
"type": "VARCHAR"
}, {
"precision": 50,
"name": "ext_tenth",
"columnDisplaySize": 50,
"type": "VARCHAR"
}, {
"precision": 50,
"name": "device_add_id",
"columnDisplaySize": 50,
"type": "VARCHAR"
}, {
"precision": 8,
"name": "today_start_value",
"columnDisplaySize": 10,
"type": "DECIMAL"
}, {
"precision": 75,
"name": "unit",
"columnDisplaySize": 75,
"type": "VARCHAR"
}],
"connection": [{
"jdbcUrl": ["jdbc:mysql://172.18.8.114:3306/Vasyslink_yag001?useSSL=false&useUnicode=true&characterEncoding=utf8"],
"table": ["tb_device_point_data_hst_01"]
}],
"splitPk": "id",
"username": "root"
},
"name": "mysqlreader"
},
"writer": {
"parameter": {
"tablesColumn": "{\"tb_device_point_data_hst_78\":[{\"key\":\"id\",\"type\":\"bigint\",\"precision\":19,\"columnDisplaySize\":20},{\"key\":\"device_id\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"point\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"hash\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"value\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"acq_time\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"ratio\",\"type\":\"float\",\"precision\":7,\"columnDisplaySize\":24},{\"key\":\"max_val\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"min_val\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"out_param\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"item_code\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"descr\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"save_hst\",\"type\":\"smallint\",\"precision\":5,\"columnDisplaySize\":6},{\"key\":\"del_flag\",\"type\":\"tinyint\",\"precision\":3,\"columnDisplaySize\":4},{\"key\":\"create_user\",\"type\":\"bigint\",\"precision\":19,\"columnDisplaySize\":20},{\"key\":\"update_user\",\"type\":\"bigint\",\"precision\":19,\"columnDisplaySize\":20},{\"key\":\"create_time\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"update_time\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"ext_first\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"ext_second\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"ext_third\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"ext_fourth\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"ext_fifth\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"ext_sixth\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"ext_seventh\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"ext_eighth\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"ext_ninth\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"ext_tenth\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"device_add_id\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647},{\"key\":\"today_start_value\",\"type\":\"decimal\",\"precision\":8,\"columnDisplaySize\":10},{\"key\":\"unit\",\"type\":\"string\",\"precision\":2147483647,\"columnDisplaySize\":2147483647}]}",
"dataSourceId": 54,
"partition": "pt",
"jdbcUrl": "jdbc:hive2://172.18.8.208:10000/Vasyslink_yag001",
"defaultFS": "hdfs://172.18.8.207:8020",
"writeMode": "overwrite",
"maxFileSize": 1073741824,
"fieldDelimiter": "\t",
"partitionType": "HOUR",
"fileType": "text",
"charsetName": "UTF-8"
},
"name": "hivewriter"
}
}],
"setting": {
"log": {
"isLogger": false
},
"errorLimit": {},
"speed": {
"bytes": 0,
"channel": 1
}
}
}
}

### How to reproduce

1、mysql数据库9000万+
2、一次性同步:mysql-hive
3、通道设置1

### Anything else

_No response_

### Version

1.12_release

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.