You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by vtygoss <vt...@126.com> on 2021/06/22 08:46:03 UTC
Flink TPC-DS 3TB BenchMark result is not good.
Hi,
I am doing performance tests on 3TB TPC-DS using flink/flink-end-to-end-tests/flink-tpcds-test module, but the test results are not good.
scenario:
tpc-ds location: hive 2.3.5
tpc-ds scala: 3TB, parquet + snappy
flink version: flink-1.12-SNAPSHOT
resource configuration: slots per task manager=5, parallesm=500, job manager memory=10GB, task manager memory=10GB, task manager number=100.
difference of my branch and community branch:
1. tpc-ds stored in hive, so i change Source from CSV to Hive.
2. I add some optimizations explicitly about join-reorder,broadcast… shown as below
3. community tpcds test main class is org.apache.flink.table.tpcds.TpcdsTestProgram, my is org.apache.flink.table.tpcds.MyTpcdsTestProgram. Both are in the attachment.
Do I make something wrong? please help to offer some advices. thanks very much!
Best Regards
```
[MyTpcdsTestProgram.java] (My Branch)
private static TableEnvironment prepareTableEnv(String sourceTablePath, Boolean useTableStats) {
// init Table Env
EnvironmentSettings environmentSettings =
EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
TableEnvironment tEnv = TableEnvironment.create(environmentSettings);
tEnv.loadModule("custom-hive", new HiveModule("2.3.5"));
String hiveSite = "/home/work/flink-1.12/conf/";
HiveCatalog catalog = new HiveCatalog("hive", "tpcds3t", hiveSite);
tEnv.registerCatalog("hive", catalog);
tEnv.useCatalog("hive");
tEnv.useDatabase("tpcds3t");
// config Optimizer parameters
tEnv.getConfig()
.getConfiguration()
.setInteger(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 500);
tEnv.getConfig()
.getConfiguration()
.setString(
ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
GlobalDataExchangeMode.POINTWISE_EDGES_PIPELINED.toString());
tEnv.getConfig()
.getConfiguration()
.setLong(
OptimizerConfigOptions.TABLE_OPTIMIZER_BROADCAST_JOIN_THRESHOLD,
10 * 1024 * 1024);
tEnv.getConfig()
.getConfiguration()
.setBoolean(OptimizerConfigOptions.TABLE_OPTIMIZER_JOIN_REORDER_ENABLED, true);
tEnv.getConfig()
.getConfiguration()
.setBoolean("table.exec.hive.infer-source-parallelism", false);
tEnv.getConfig().getConfiguration().setString("table.exec.sink.not-null-enforcer", "drop");
tEnv.getConfig()
.getConfiguration()
.setBoolean("table.optimizer.join-reorder-enabled", true);
tEnv.getConfig()
.getConfiguration()
.setInteger("table.optimizer.distinct-agg.split.bucket-num", 10240);
tEnv.getConfig()
.getConfiguration()
.setBoolean("table.optimizer.distinct-agg.split.enabled", true);
tEnv.getConfig()
.getConfiguration()
.setInteger("table.optimizer.join.broadcast-threshold", 256 * 1024 * 1024);
tEnv.getConfig()
.getConfiguration()
.setBoolean("table.optimizer.reuse-sub-plan-enabled", true);
tEnv.getConfig()
.getConfiguration()
.setBoolean("table.optimizer.reuse-source-enabled", true);
tEnv.getConfig()
.getConfiguration()
.setBoolean("table.optimizer.source.predicate-pushdown-enabled", true);
/* STREAMING
tEnv.getConfig()
.getConfiguration()
.setInteger("table.exec.async-lookup.buffer-capacity", 10000);
tEnv.getConfig().getConfiguration().setBoolean("table.exec.mini-batch.enabled", true);
tEnv.getConfig()
.getConfiguration()
.setString("table.exec.mini-batch.allow-latency", "1 min");
tEnv.getConfig().getConfiguration().setInteger("table.exec.mini-batch.size", 10000);*/
// register statistics info
if (useTableStats) {
TpcdsStatsProvider.registerTpcdsStats(tEnv);
}
return tEnv;
}
```
```
TpcdsTestProgram.java(community branch)
private static TableEnvironment prepareTableEnv(String sourceTablePath, Boolean useTableStats) {
// init Table Env
EnvironmentSettings environmentSettings =
EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
TableEnvironment tEnv = TableEnvironment.create(environmentSettings);
// config Optimizer parameters
tEnv.getConfig()
.getConfiguration()
.setInteger(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 4);
tEnv.getConfig()
.getConfiguration()
.setString(
ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
GlobalDataExchangeMode.POINTWISE_EDGES_PIPELINED.toString());
tEnv.getConfig()
.getConfiguration()
.setLong(
OptimizerConfigOptions.TABLE_OPTIMIZER_BROADCAST_JOIN_THRESHOLD,
10 * 1024 * 1024);
tEnv.getConfig()
.getConfiguration()
.setBoolean(OptimizerConfigOptions.TABLE_OPTIMIZER_JOIN_REORDER_ENABLED, true);
// register TPC-DS tables
TPCDS_TABLES.forEach(
table -> {
TpcdsSchema schema = TpcdsSchemaProvider.getTableSchema(table);
CsvTableSource.Builder builder = CsvTableSource.builder();
builder.path(sourceTablePath + FILE_SEPARATOR + table + DATA_SUFFIX);
for (int i = 0; i < schema.getFieldNames().size(); i++) {
builder.field(
schema.getFieldNames().get(i),
TypeConversions.fromDataTypeToLegacyInfo(
schema.getFieldTypes().get(i)));
}
builder.fieldDelimiter(COL_DELIMITER);
builder.emptyColumnAsNull();
builder.lineDelimiter("\n");
CsvTableSource tableSource = builder.build();
ConnectorCatalogTable catalogTable =
ConnectorCatalogTable.source(tableSource, true);
tEnv.getCatalog(tEnv.getCurrentCatalog())
.ifPresent(
catalog -> {
try {
catalog.createTable(
new ObjectPath(
tEnv.getCurrentDatabase(), table),
catalogTable,
false);
} catch (Exception e) {
throw new RuntimeException(e);
}
});
});
// register statistics info
if (useTableStats) {
TpcdsStatsProvider.registerTpcdsStats(tEnv);
}
return tEnv;
}
```
e.g.
query 1 takes about 40min to execute(standard spark-2.3.0 token 13 second, spark-2.3.0 with customed Adaptive Query Execution feature token 5.99s), flink “Explan plan“ of query 1 is in the attachment.
```
query1.sql
with customer_total_return as
(select sr_customer_sk as ctr_customer_sk
,sr_store_sk as ctr_store_sk
,sum(sr_return_amt) as ctr_total_return
from store_returns
,date_dim
where sr_returned_date_sk = d_date_sk
and d_year = 2000
group by sr_customer_sk
,sr_store_sk)
select c_customer_id
from customer_total_return ctr1
,store
,customer
where ctr1.ctr_total_return > (select avg(ctr_total_return)*1.2
from customer_total_return ctr2
where ctr1.ctr_store_sk = ctr2.ctr_store_sk)
and s_store_sk = ctr1.ctr_store_sk
and s_state = 'TN'
and ctr1.ctr_customer_sk = c_customer_sk
order by c_customer_id
limit 100
```
stages:
job configurations:
JM memory:
TM memory:
Re: Flink TPC-DS 3TB BenchMark result is not good.
Posted by Jingsong Li <ji...@gmail.com>.
Thanks Yingjie for pinging me.
Hi vtygoss,
Leonard is right, maybe you are using the wrong statistics information.
This caused the optimizer to select the **BROADCAST JOIN**
incorrectly. Unfortunately, Flink needs to broadcast a huge amount of data,
even gigabytes. This is really the performance killer.
I think you can:
- analyze your tables in Hive as Leonard said.
- Or just remove "TpcdsStatsProvider.registerTpcdsStats(tEnv)"
And I see your code: "table.optimizer.join.broadcast-threshold: 256 * 1024
* 1024".
I think this threshold is too large. More than 10MB is not recommended.
Best,
Jingsong
On Wed, Jun 23, 2021 at 11:08 AM Leonard Xu <xb...@gmail.com> wrote:
> Hi, vtygoss
>
> Thanks for the detail report, a quick reply as I wrote the
> * org.apache.flink.table.tpcds.TpcdsTestProgram* in community, I guess
> you missed *table statistics information.*
>
> The* table statistics information* used in the TPC-DS e2e tests is
> constant for 1GB verification data set, I wrote this test for checking
> Flink Batch SQL works well for every PR as CI test rather than checking the
> performance. Please see
> *org.apache.flink.table.tpcds.stats.TpcdsStatsProvider*.
>
> The *table statistics information *will be used by planner(CBO
> optimizer) to optimize the sql plan, the incorrect *table statistics
> information *even lead to the wrong plan and sql job may run unexpectedly.
>
> Thus if you want to run for 3TB TPC-DS tests, you should use the
> corresponding *table statistics information *for your test data set, you
> can obtain the table statistics information by analyze your tables in Hive.
>
> Best,
> Leonard
>
> 在 2021年6月23日，10:42，Yingjie Cao <ke...@gmail.com> 写道：
>
> Hi,
>
> I also have some experience of running TPC-DS benchmark with Flink (10T
> scale). But the data shuffle amount of Q1 has a really big difference with
> the numbers in the picture you shared. I am not sure what is going on,
> maybe you missed something? I attached the numbers of Q1 in my test (also
> with 500 max parallelism, though I used Flink version 1.13 instead of
> 1.12), the running time is 20s for 10T TPC-DS.
>
> There are some points I known which may influence the test results, hope
> these can help you:
> 1. Shuffle data compression. (Disabled in Flink by default, can be enabled
> by setting taskmanager.network.blocking-shuffle.compression.enabled to
> true);
> 2. Blocking shuffle type used. See [1] for more information. (To used
> sort-shuffle, the minimum version is 1.13);
> 3. Memory configuration, including network and managed memory size.
> 4. Sql optimization configuration. (I am not familiar with this,
> maybe @Jingsong Li has more knowledge about that).
>
> BTW, could you please share more information? For example, how many nodes
> in your cluster? Which type of disk do you use, SSD or HDD? How many
> available cores, disks and memory of each node? Could you also share the
> numbers of shuffle write and shuffle read of all stages of Spark?
>
> [1]
> https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/ops/batch/blocking_shuffle/
>
> <image.png>
>
> Best,
> Yingjie
>
> vtygoss <vt...@126.com> 于2021年6月22日周二 下午4:46写道：
>
>> Hi,
>>
>> I am doing performance tests on 3TB TPC-DS using
>> flink/flink-end-to-end-tests/flink-tpcds-test module, but the test results
>> are not good.
>>
>> scenario:
>> tpc-ds location: hive 2.3.5
>> tpc-ds scala: 3TB, parquet + snappy
>> flink version: flink-1.12-SNAPSHOT
>> resource configuration: slots per task manager=5, parallesm=500, job
>> manager memory=10GB, task manager memory=10GB, task manager number=100.
>>
>> difference of my branch and community branch:
>> 1. tpc-ds stored in hive, so i change Source from CSV to Hive.
>> 2. I add some optimizations explicitly about join-reorder,broadcast…
>> shown as below
>> 3. community tpcds test main class is org.apache.flink.table.tpcds.
>> TpcdsTestProgram, my is org.apache.flink.table.tpcds.My
>> <http://org.apache.flink.table.tpcds.my/>TpcdsTestProgram. Both are in
>> the attachment.
>>
>> Do I make something wrong? please help to offer some advices. thanks very
>> much!
>>
>> Best Regards
>>
>>
>> ```
>> [MyTpcdsTestProgram.java] (My Branch)
>>
>> private static TableEnvironment prepareTableEnv(String sourceTablePath, Boolean useTableStats) {
>> // init Table Env
>> EnvironmentSettings environmentSettings =
>> EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
>>
>> TableEnvironment tEnv = TableEnvironment.create(environmentSettings);
>>
>> tEnv.loadModule("custom-hive", new HiveModule("2.3.5"));
>> String hiveSite = "/home/work/flink-1.12/conf/";
>> HiveCatalog catalog = new HiveCatalog("hive", "tpcds3t", hiveSite);
>> tEnv.registerCatalog("hive", catalog);
>> tEnv.useCatalog("hive");
>> tEnv.useDatabase("tpcds3t");
>> // config Optimizer parameters
>> tEnv.getConfig()
>> .getConfiguration()
>> .setInteger(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 500);
>> tEnv.getConfig()
>> .getConfiguration()
>> .setString(
>> ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
>> GlobalDataExchangeMode.POINTWISE_EDGES_PIPELINED.toString());
>> tEnv.getConfig()
>> .getConfiguration()
>> .setLong(
>> OptimizerConfigOptions.TABLE_OPTIMIZER_BROADCAST_JOIN_THRESHOLD,
>> 10 * 1024 * 1024);
>> tEnv.getConfig()
>> .getConfiguration()
>> .setBoolean(OptimizerConfigOptions.TABLE_OPTIMIZER_JOIN_REORDER_ENABLED, true);
>>
>> tEnv.getConfig()
>> .getConfiguration()
>> .setBoolean("table.exec.hive.infer-source-parallelism", false);
>> tEnv.getConfig().getConfiguration().setString("table.exec.sink.not-null-enforcer", "drop");
>>
>> tEnv.getConfig()
>> .getConfiguration()
>> .setBoolean("table.optimizer.join-reorder-enabled", true);
>> tEnv.getConfig()
>> .getConfiguration()
>> .setInteger("table.optimizer.distinct-agg.split.bucket-num", 10240);
>> tEnv.getConfig()
>> .getConfiguration()
>> .setBoolean("table.optimizer.distinct-agg.split.enabled", true);
>> tEnv.getConfig()
>> .getConfiguration()
>> .setInteger("table.optimizer.join.broadcast-threshold", 256 * 1024 * 1024);
>> tEnv.getConfig()
>> .getConfiguration()
>> .setBoolean("table.optimizer.reuse-sub-plan-enabled", true);
>> tEnv.getConfig()
>> .getConfiguration()
>> .setBoolean("table.optimizer.reuse-source-enabled", true);
>> tEnv.getConfig()
>> .getConfiguration()
>> .setBoolean("table.optimizer.source.predicate-pushdown-enabled", true);
>>
>> /* STREAMING
>> tEnv.getConfig()
>> .getConfiguration()
>> .setInteger("table.exec.async-lookup.buffer-capacity", 10000);
>> tEnv.getConfig().getConfiguration().setBoolean("table.exec.mini-batch.enabled", true);
>> tEnv.getConfig()
>> .getConfiguration()
>> .setString("table.exec.mini-batch.allow-latency", "1 min");
>> tEnv.getConfig().getConfiguration().setInteger("table.exec.mini-batch.size", 10000);*/
>> // register statistics info
>> if (useTableStats) {
>> TpcdsStatsProvider.registerTpcdsStats(tEnv);
>> }
>> return tEnv;
>> }
>>
>> ```
>>
>>
>> ```
>> TpcdsTestProgram.java(community branch)
>>
>> private static TableEnvironment prepareTableEnv(String sourceTablePath, Boolean useTableStats) {
>> // init Table Env
>> EnvironmentSettings environmentSettings =
>> EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
>> TableEnvironment tEnv = TableEnvironment.create(environmentSettings);
>>
>> // config Optimizer parameters
>> tEnv.getConfig()
>> .getConfiguration()
>> .setInteger(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 4);
>> tEnv.getConfig()
>> .getConfiguration()
>> .setString(
>> ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
>> GlobalDataExchangeMode.POINTWISE_EDGES_PIPELINED.toString());
>> tEnv.getConfig()
>> .getConfiguration()
>> .setLong(
>> OptimizerConfigOptions.TABLE_OPTIMIZER_BROADCAST_JOIN_THRESHOLD,
>> 10 * 1024 * 1024);
>> tEnv.getConfig()
>> .getConfiguration()
>> .setBoolean(OptimizerConfigOptions.TABLE_OPTIMIZER_JOIN_REORDER_ENABLED, true);
>>
>> // register TPC-DS tables
>> TPCDS_TABLES.forEach(
>> table -> {
>> TpcdsSchema schema = TpcdsSchemaProvider.getTableSchema(table);
>> CsvTableSource.Builder builder = CsvTableSource.builder();
>> builder.path(sourceTablePath + FILE_SEPARATOR + table + DATA_SUFFIX);
>> for (int i = 0; i < schema.getFieldNames().size(); i++) {
>> builder.field(
>> schema.getFieldNames().get(i),
>> TypeConversions.fromDataTypeToLegacyInfo(
>> schema.getFieldTypes().get(i)));
>> }
>> builder.fieldDelimiter(COL_DELIMITER);
>> builder.emptyColumnAsNull();
>> builder.lineDelimiter("\n");
>> CsvTableSource tableSource = builder.build();
>> ConnectorCatalogTable catalogTable =
>> ConnectorCatalogTable.source(tableSource, true);
>> tEnv.getCatalog(tEnv.getCurrentCatalog())
>> .ifPresent(
>> catalog -> {
>> try {
>> catalog.createTable(
>> new ObjectPath(
>> tEnv.getCurrentDatabase(), table),
>> catalogTable,
>> false);
>> } catch (Exception e) {
>> throw new RuntimeException(e);
>> }
>> });
>> });
>> // register statistics info
>> if (useTableStats) {
>> TpcdsStatsProvider.registerTpcdsStats(tEnv);
>> }
>> return tEnv;
>> }
>>
>> ```
>>
>>
>>
>>
>> e.g.
>> query 1 takes about 40min to execute(standard spark-2.3.0 token 13
>> second, spark-2.3.0 with customed Adaptive Query Execution feature token
>> 5.99s), flink “Explan plan“ of query 1 is in the attachment.
>>
>>
>> ```
>> query1.sql
>>
>> with customer_total_return as
>> (select sr_customer_sk as ctr_customer_sk
>> ,sr_store_sk as ctr_store_sk
>> ,sum(sr_return_amt) as ctr_total_return
>> from store_returns
>> ,date_dim
>> where sr_returned_date_sk = d_date_sk
>> and d_year = 2000
>> group by sr_customer_sk
>> ,sr_store_sk)
>> select c_customer_id
>> from customer_total_return ctr1
>> ,store
>> ,customer
>> where ctr1.ctr_total_return > (select avg(ctr_total_return)*1.2
>> from customer_total_return ctr2
>> where ctr1.ctr_store_sk = ctr2.ctr_store_sk)
>> and s_store_sk = ctr1.ctr_store_sk
>> and s_state = 'TN'
>> and ctr1.ctr_customer_sk = c_customer_sk
>> order by c_customer_id
>> limit 100
>>
>> ```
>>
>>
>>
>> stages:
>> <881f3241-29c4-4aa0-9c08-43560817a9eb.png>
>>
>> job configurations:
>> <41a1da0b-8655-447f-8b2c-c2119c3ee1a3.png>
>>
>> JM memory:
>> <1562c554-c765-47b8-b02e-bd71d1abb6d1.png>
>>
>> TM memory:
>> <f4c3eb06-9daf-4b48-8a65-e700d2b9c5ee.png>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>
--
Best, Jingsong Lee
Re: Flink TPC-DS 3TB BenchMark result is not good.
Posted by Leonard Xu <xb...@gmail.com>.
Hi, vtygoss
Thanks for the detail report, a quick reply as I wrote the org.apache.flink.table.tpcds.TpcdsTestProgram in community, I guess you missed table statistics information.
The table statistics information used in the TPC-DS e2e tests is constant for 1GB verification data set, I wrote this test for checking Flink Batch SQL works well for every PR as CI test rather than checking the performance. Please see org.apache.flink.table.tpcds.stats.TpcdsStatsProvider.
The table statistics information will be used by planner(CBO optimizer) to optimize the sql plan, the incorrect table statistics information even lead to the wrong plan and sql job may run unexpectedly.
Thus if you want to run for 3TB TPC-DS tests, you should use the corresponding table statistics information for your test data set, you can obtain the table statistics information by analyze your tables in Hive.
Best,
Leonard
> 在 2021年6月23日，10:42，Yingjie Cao <ke...@gmail.com> 写道：
>
> Hi,
>
> I also have some experience of running TPC-DS benchmark with Flink (10T scale). But the data shuffle amount of Q1 has a really big difference with the numbers in the picture you shared. I am not sure what is going on, maybe you missed something? I attached the numbers of Q1 in my test (also with 500 max parallelism, though I used Flink version 1.13 instead of 1.12), the running time is 20s for 10T TPC-DS.
>
> There are some points I known which may influence the test results, hope these can help you:
> 1. Shuffle data compression. (Disabled in Flink by default, can be enabled by setting taskmanager.network.blocking-shuffle.compression.enabled to true);
> 2. Blocking shuffle type used. See [1] for more information. (To used sort-shuffle, the minimum version is 1.13);
> 3. Memory configuration, including network and managed memory size.
> 4. Sql optimization configuration. (I am not familiar with this, maybe @Jingsong Li has more knowledge about that).
>
> BTW, could you please share more information? For example, how many nodes in your cluster? Which type of disk do you use, SSD or HDD? How many available cores, disks and memory of each node? Could you also share the numbers of shuffle write and shuffle read of all stages of Spark?
>
> [1] https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/ops/batch/blocking_shuffle/ <https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/ops/batch/blocking_shuffle/>
>
> <image.png>
>
> Best,
> Yingjie
>
> vtygoss <vtygoss@126.com <ma...@126.com>> 于2021年6月22日周二 下午4:46写道：
> Hi,
>
> I am doing performance tests on 3TB TPC-DS using flink/flink-end-to-end-tests/flink-tpcds-test module, but the test results are not good.
>
> scenario:
> tpc-ds location: hive 2.3.5
> tpc-ds scala: 3TB, parquet + snappy
> flink version: flink-1.12-SNAPSHOT
> resource configuration: slots per task manager=5, parallesm=500, job manager memory=10GB, task manager memory=10GB, task manager number=100.
>
> difference of my branch and community branch:
> 1. tpc-ds stored in hive, so i change Source from CSV to Hive.
> 2. I add some optimizations explicitly about join-reorder,broadcast… shown as below
> 3. community tpcds test main class is org.apache.flink.table.tpcds.TpcdsTestProgram, my is org.apache.flink.table.tpcds.My <http://org.apache.flink.table.tpcds.my/>TpcdsTestProgram. Both are in the attachment.
>
> Do I make something wrong? please help to offer some advices. thanks very much!
>
> Best Regards
>
>
> ```
> [MyTpcdsTestProgram.java] (My Branch)
> private static TableEnvironment prepareTableEnv(String sourceTablePath, Boolean useTableStats) {
> // init Table Env
> EnvironmentSettings environmentSettings =
> EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
>
> TableEnvironment tEnv = TableEnvironment.create(environmentSettings);
>
> tEnv.loadModule("custom-hive", new HiveModule("2.3.5"));
> String hiveSite = "/home/work/flink-1.12/conf/";
> HiveCatalog catalog = new HiveCatalog("hive", "tpcds3t", hiveSite);
> tEnv.registerCatalog("hive", catalog);
> tEnv.useCatalog("hive");
> tEnv.useDatabase("tpcds3t");
> // config Optimizer parameters
> tEnv.getConfig()
> .getConfiguration()
> .setInteger(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 500);
> tEnv.getConfig()
> .getConfiguration()
> .setString(
> ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
> GlobalDataExchangeMode.POINTWISE_EDGES_PIPELINED.toString());
> tEnv.getConfig()
> .getConfiguration()
> .setLong(
> OptimizerConfigOptions.TABLE_OPTIMIZER_BROADCAST_JOIN_THRESHOLD,
> 10 * 1024 * 1024);
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean(OptimizerConfigOptions.TABLE_OPTIMIZER_JOIN_REORDER_ENABLED, true);
>
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean("table.exec.hive.infer-source-parallelism", false);
> tEnv.getConfig().getConfiguration().setString("table.exec.sink.not-null-enforcer", "drop");
>
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean("table.optimizer.join-reorder-enabled", true);
> tEnv.getConfig()
> .getConfiguration()
> .setInteger("table.optimizer.distinct-agg.split.bucket-num", 10240);
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean("table.optimizer.distinct-agg.split.enabled", true);
> tEnv.getConfig()
> .getConfiguration()
> .setInteger("table.optimizer.join.broadcast-threshold", 256 * 1024 * 1024);
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean("table.optimizer.reuse-sub-plan-enabled", true);
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean("table.optimizer.reuse-source-enabled", true);
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean("table.optimizer.source.predicate-pushdown-enabled", true);
>
> /* STREAMING
> tEnv.getConfig()
> .getConfiguration()
> .setInteger("table.exec.async-lookup.buffer-capacity", 10000);
> tEnv.getConfig().getConfiguration().setBoolean("table.exec.mini-batch.enabled", true);
> tEnv.getConfig()
> .getConfiguration()
> .setString("table.exec.mini-batch.allow-latency", "1 min");
> tEnv.getConfig().getConfiguration().setInteger("table.exec.mini-batch.size", 10000);*/
> // register statistics info
> if (useTableStats) {
> TpcdsStatsProvider.registerTpcdsStats(tEnv);
> }
> return tEnv;
> }
> ```
>
>
> ```
> TpcdsTestProgram.java(community branch)
> private static TableEnvironment prepareTableEnv(String sourceTablePath, Boolean useTableStats) {
> // init Table Env
> EnvironmentSettings environmentSettings =
> EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
> TableEnvironment tEnv = TableEnvironment.create(environmentSettings);
>
> // config Optimizer parameters
> tEnv.getConfig()
> .getConfiguration()
> .setInteger(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 4);
> tEnv.getConfig()
> .getConfiguration()
> .setString(
> ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
> GlobalDataExchangeMode.POINTWISE_EDGES_PIPELINED.toString());
> tEnv.getConfig()
> .getConfiguration()
> .setLong(
> OptimizerConfigOptions.TABLE_OPTIMIZER_BROADCAST_JOIN_THRESHOLD,
> 10 * 1024 * 1024);
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean(OptimizerConfigOptions.TABLE_OPTIMIZER_JOIN_REORDER_ENABLED, true);
>
> // register TPC-DS tables
> TPCDS_TABLES.forEach(
> table -> {
> TpcdsSchema schema = TpcdsSchemaProvider.getTableSchema(table);
> CsvTableSource.Builder builder = CsvTableSource.builder();
> builder.path(sourceTablePath + FILE_SEPARATOR + table + DATA_SUFFIX);
> for (int i = 0; i < schema.getFieldNames().size(); i++) {
> builder.field(
> schema.getFieldNames().get(i),
> TypeConversions.fromDataTypeToLegacyInfo(
> schema.getFieldTypes().get(i)));
> }
> builder.fieldDelimiter(COL_DELIMITER);
> builder.emptyColumnAsNull();
> builder.lineDelimiter("\n");
> CsvTableSource tableSource = builder.build();
> ConnectorCatalogTable catalogTable =
> ConnectorCatalogTable.source(tableSource, true);
> tEnv.getCatalog(tEnv.getCurrentCatalog())
> .ifPresent(
> catalog -> {
> try {
> catalog.createTable(
> new ObjectPath(
> tEnv.getCurrentDatabase(), table),
> catalogTable,
> false);
> } catch (Exception e) {
> throw new RuntimeException(e);
> }
> });
> });
> // register statistics info
> if (useTableStats) {
> TpcdsStatsProvider.registerTpcdsStats(tEnv);
> }
> return tEnv;
> }
> ```
>
>
>
>
> e.g.
> query 1 takes about 40min to execute(standard spark-2.3.0 token 13 second, spark-2.3.0 with customed Adaptive Query Execution feature token 5.99s), flink “Explan plan“ of query 1 is in the attachment.
>
>
> ```
> query1.sql
> with customer_total_return as
> (select sr_customer_sk as ctr_customer_sk
> ,sr_store_sk as ctr_store_sk
> ,sum(sr_return_amt) as ctr_total_return
> from store_returns
> ,date_dim
> where sr_returned_date_sk = d_date_sk
> and d_year = 2000
> group by sr_customer_sk
> ,sr_store_sk)
> select c_customer_id
> from customer_total_return ctr1
> ,store
> ,customer
> where ctr1.ctr_total_return > (select avg(ctr_total_return)*1.2
> from customer_total_return ctr2
> where ctr1.ctr_store_sk = ctr2.ctr_store_sk)
> and s_store_sk = ctr1.ctr_store_sk
> and s_state = 'TN'
> and ctr1.ctr_customer_sk = c_customer_sk
> order by c_customer_id
> limit 100
> ```
>
>
>
> stages:
> <881f3241-29c4-4aa0-9c08-43560817a9eb.png>
>
> job configurations:
> <41a1da0b-8655-447f-8b2c-c2119c3ee1a3.png>
>
> JM memory:
> <1562c554-c765-47b8-b02e-bd71d1abb6d1.png>
>
> TM memory:
> <f4c3eb06-9daf-4b48-8a65-e700d2b9c5ee.png>
>
>
>
>
>
>
>
>
>
>
>
>
Re: Flink TPC-DS 3TB BenchMark result is not good.
Posted by Yingjie Cao <ke...@gmail.com>.
Hi,
I also have some experience of running TPC-DS benchmark with Flink (10T
scale). But the data shuffle amount of Q1 has a really big difference with
the numbers in the picture you shared. I am not sure what is going on,
maybe you missed something? I attached the numbers of Q1 in my test (also
with 500 max parallelism, though I used Flink version 1.13 instead of
1.12), the running time is 20s for 10T TPC-DS.
There are some points I known which may influence the test results, hope
these can help you:
1. Shuffle data compression. (Disabled in Flink by default, can be enabled
by setting taskmanager.network.blocking-shuffle.compression.enabled to
true);
2. Blocking shuffle type used. See [1] for more information. (To used
sort-shuffle, the minimum version is 1.13);
3. Memory configuration, including network and managed memory size.
4. Sql optimization configuration. (I am not familiar with this,
maybe @Jingsong Li has more knowledge about that).
BTW, could you please share more information? For example, how many nodes
in your cluster? Which type of disk do you use, SSD or HDD? How many
available cores, disks and memory of each node? Could you also share the
numbers of shuffle write and shuffle read of all stages of Spark?
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/ops/batch/blocking_shuffle/
[image: image.png]
Best,
Yingjie
vtygoss <vt...@126.com> 于2021年6月22日周二 下午4:46写道：
> Hi,
>
>
> I am doing performance tests on 3TB TPC-DS using
> flink/flink-end-to-end-tests/flink-tpcds-test module, but the test results
> are not good.
>
>
> scenario:
>
> tpc-ds location: hive 2.3.5
>
> tpc-ds scala: 3TB, parquet + snappy
>
> flink version: flink-1.12-SNAPSHOT
>
> resource configuration: slots per task manager=5, parallesm=500, job
> manager memory=10GB, task manager memory=10GB, task manager number=100.
>
>
> difference of my branch and community branch:
>
> 1. tpc-ds stored in hive, so i change Source from CSV to Hive.
>
> 2. I add some optimizations explicitly about join-reorder,broadcast…
> shown as below
>
> 3. community tpcds test main class is org.apache.flink.table.tpcds.
> TpcdsTestProgram, my is org.apache.flink.table.tpcds.MyTpcdsTestProgram. Both
> are in the attachment.
>
>
> Do I make something wrong? please help to offer some advices. thanks very
> much!
>
>
> Best Regards
>
>
>
> ```
> [MyTpcdsTestProgram.java] (My Branch)
>
> private static TableEnvironment prepareTableEnv(String sourceTablePath, Boolean useTableStats) {
> // init Table Env
> EnvironmentSettings environmentSettings =
> EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
>
> TableEnvironment tEnv = TableEnvironment.create(environmentSettings);
>
> tEnv.loadModule("custom-hive", new HiveModule("2.3.5"));
> String hiveSite = "/home/work/flink-1.12/conf/";
> HiveCatalog catalog = new HiveCatalog("hive", "tpcds3t", hiveSite);
> tEnv.registerCatalog("hive", catalog);
> tEnv.useCatalog("hive");
> tEnv.useDatabase("tpcds3t");
> // config Optimizer parameters
> tEnv.getConfig()
> .getConfiguration()
> .setInteger(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 500);
> tEnv.getConfig()
> .getConfiguration()
> .setString(
> ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
> GlobalDataExchangeMode.POINTWISE_EDGES_PIPELINED.toString());
> tEnv.getConfig()
> .getConfiguration()
> .setLong(
> OptimizerConfigOptions.TABLE_OPTIMIZER_BROADCAST_JOIN_THRESHOLD,
> 10 * 1024 * 1024);
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean(OptimizerConfigOptions.TABLE_OPTIMIZER_JOIN_REORDER_ENABLED, true);
>
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean("table.exec.hive.infer-source-parallelism", false);
> tEnv.getConfig().getConfiguration().setString("table.exec.sink.not-null-enforcer", "drop");
>
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean("table.optimizer.join-reorder-enabled", true);
> tEnv.getConfig()
> .getConfiguration()
> .setInteger("table.optimizer.distinct-agg.split.bucket-num", 10240);
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean("table.optimizer.distinct-agg.split.enabled", true);
> tEnv.getConfig()
> .getConfiguration()
> .setInteger("table.optimizer.join.broadcast-threshold", 256 * 1024 * 1024);
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean("table.optimizer.reuse-sub-plan-enabled", true);
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean("table.optimizer.reuse-source-enabled", true);
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean("table.optimizer.source.predicate-pushdown-enabled", true);
>
> /* STREAMING
> tEnv.getConfig()
> .getConfiguration()
> .setInteger("table.exec.async-lookup.buffer-capacity", 10000);
> tEnv.getConfig().getConfiguration().setBoolean("table.exec.mini-batch.enabled", true);
> tEnv.getConfig()
> .getConfiguration()
> .setString("table.exec.mini-batch.allow-latency", "1 min");
> tEnv.getConfig().getConfiguration().setInteger("table.exec.mini-batch.size", 10000);*/
> // register statistics info
> if (useTableStats) {
> TpcdsStatsProvider.registerTpcdsStats(tEnv);
> }
> return tEnv;
> }
>
> ```
>
>
> ```
> TpcdsTestProgram.java(community branch)
>
> private static TableEnvironment prepareTableEnv(String sourceTablePath, Boolean useTableStats) {
> // init Table Env
> EnvironmentSettings environmentSettings =
> EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
> TableEnvironment tEnv = TableEnvironment.create(environmentSettings);
>
> // config Optimizer parameters
> tEnv.getConfig()
> .getConfiguration()
> .setInteger(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 4);
> tEnv.getConfig()
> .getConfiguration()
> .setString(
> ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
> GlobalDataExchangeMode.POINTWISE_EDGES_PIPELINED.toString());
> tEnv.getConfig()
> .getConfiguration()
> .setLong(
> OptimizerConfigOptions.TABLE_OPTIMIZER_BROADCAST_JOIN_THRESHOLD,
> 10 * 1024 * 1024);
> tEnv.getConfig()
> .getConfiguration()
> .setBoolean(OptimizerConfigOptions.TABLE_OPTIMIZER_JOIN_REORDER_ENABLED, true);
>
> // register TPC-DS tables
> TPCDS_TABLES.forEach(
> table -> {
> TpcdsSchema schema = TpcdsSchemaProvider.getTableSchema(table);
> CsvTableSource.Builder builder = CsvTableSource.builder();
> builder.path(sourceTablePath + FILE_SEPARATOR + table + DATA_SUFFIX);
> for (int i = 0; i < schema.getFieldNames().size(); i++) {
> builder.field(
> schema.getFieldNames().get(i),
> TypeConversions.fromDataTypeToLegacyInfo(
> schema.getFieldTypes().get(i)));
> }
> builder.fieldDelimiter(COL_DELIMITER);
> builder.emptyColumnAsNull();
> builder.lineDelimiter("\n");
> CsvTableSource tableSource = builder.build();
> ConnectorCatalogTable catalogTable =
> ConnectorCatalogTable.source(tableSource, true);
> tEnv.getCatalog(tEnv.getCurrentCatalog())
> .ifPresent(
> catalog -> {
> try {
> catalog.createTable(
> new ObjectPath(
> tEnv.getCurrentDatabase(), table),
> catalogTable,
> false);
> } catch (Exception e) {
> throw new RuntimeException(e);
> }
> });
> });
> // register statistics info
> if (useTableStats) {
> TpcdsStatsProvider.registerTpcdsStats(tEnv);
> }
> return tEnv;
> }
>
> ```
>
>
>
>
> e.g.
>
> query 1 takes about 40min to execute(standard spark-2.3.0 token 13 second,
> spark-2.3.0 with customed Adaptive Query Execution feature token 5.99s),
> flink “Explan plan“ of query 1 is in the attachment.
>
>
>
> ```
>
> query1.sql
>
> with customer_total_return as
> (select sr_customer_sk as ctr_customer_sk
> ,sr_store_sk as ctr_store_sk
> ,sum(sr_return_amt) as ctr_total_return
> from store_returns
> ,date_dim
> where sr_returned_date_sk = d_date_sk
> and d_year = 2000
> group by sr_customer_sk
> ,sr_store_sk)
> select c_customer_id
> from customer_total_return ctr1
> ,store
> ,customer
> where ctr1.ctr_total_return > (select avg(ctr_total_return)*1.2
> from customer_total_return ctr2
> where ctr1.ctr_store_sk = ctr2.ctr_store_sk)
> and s_store_sk = ctr1.ctr_store_sk
> and s_state = 'TN'
> and ctr1.ctr_customer_sk = c_customer_sk
> order by c_customer_id
> limit 100
>
> ```
>
>
>
>
> stages:
>
>
> job configurations:
>
>
> JM memory:
>
>
> TM memory:
>
>
>
>
>
>
>
>
>
>
>
>
>
>