Duplicated data while replication

1,299 views
Skip to first unread message

Kost Kost

unread,
Sep 4, 2017, 11:57:45 AM9/4/17
to ClickHouse
Hello!

I got a well known problem with duplicating data on Distributed and Replicated table but I can not solve it any way...
There are two test nodes replicating each other with the following config on each (only macros section differes on the second node) https://pastebin.com/E1YAypgm.
Tables created as follows:
CREATE TABLE tesd (...) ENGINE ReplicatedMergeTree('/clickhouse/tables/{slicer_shard}/default/tesd', '{replica}', date, (siteId, date, docUuid), 8192);
CREATE TABLE tesd_all AS tesd ENGINE = Distributed(slicer, default, tesd, cityHash64(docUuid));

Making one insert into tesd_all table leads to 2 rows on select operation.

Thank you in advance!

Николай Кочетов

unread,
Sep 5, 2017, 9:44:12 AM9/5/17
to ClickHouse
Hello!

As I can see, in your configuration both shards look at the same tables. So, during reading the same data was read from the same table twice.

Probably, you want to achieve cross-replication. Working example can be found here (but for 3 nodes).

Basically, you need to create 2 tables on each node (in different databases), and set up cluster and replica number manually. For example:
Node 1:
CREATE DATABASE shard_0;
CREATE DATABASE shard_1;
CREATE TABLE shard_0.tesd (...) ENGINE ReplicatedMergeTree('/clickhouse/tables/slicer00/default/tesd', 'server0', date, (siteId, date, docUuid), 8192);
CREATE TABLE shard_1.tesd (...) ENGINE ReplicatedMergeTree('/clickhouse/tables/slicer01/default/tesd', 'server0', date, (siteId, date, docUuid), 8192);
CREATE TABLE tesd_all AS tesd ENGINE = Distributed(slicer, '', tesd, cityHash64(docUuid));

Node 1:
CREATE DATABASE shard_0;
CREATE DATABASE shard_1;
CREATE TABLE shard_0.tesd (...) ENGINE ReplicatedMergeTree('/clickhouse/tables/slicer00/default/tesd', 'server1', date, (siteId, date, docUuid), 8192);
CREATE TABLE shard_1.tesd (...) ENGINE ReplicatedMergeTree('/clickhouse/tables/slicer01/default/tesd', 'server1', date, (siteId, date, docUuid), 8192);
CREATE TABLE tesd_all AS tesd ENGINE = Distributed(slicer, '', tesd, cityHash64(docUuid));

Also, you need to add:
<default_database>shard_0</default_database>
or
<default_database>shard_1</default_database>
in cluster configuration for each replica.

понедельник, 4 сентября 2017 г., 18:57:45 UTC+3 пользователь Kost Kost написал:

Kost Kost

unread,
Sep 6, 2017, 8:04:15 AM9/6/17
to ClickHouse

Hello and thank you for the answer!


I've followed all recommendations you gave and got the same result...

Also I have tried to setup only one shard with two replicas and it surely gave correct result with one row from select request... but i need a real distributed schema both on select and insert requests.


Here is all the steps I make on my test nodes (red mark means different value on different servers)


Minor changes in root clickhouse config /etc/clickhouse-server/config.xml on both nodes:

<default_database>slicer00</default_database>

<listen_host>::</listen_host>

<include_from>/etc/clickhouse-server/my.xml</include_from>


Same /etc/clickhouse-server/my.xml on both nodes:

<yandex>

<clickhouse_remote_servers>

<slicer>

<shard>

<internal_replication>true</internal_replication>

<replica>

<host>10.246.1.115</host>

<port>9000</port>

</replica>

<replica>

<host>10.246.1.93</host>

<port>9000</port>

</replica>

</shard>

<shard>

<internal_replication>true</internal_replication>

<replica>

<host>10.246.1.115</host>

<port>9000</port>

</replica>

<replica>

<host>10.246.1.93</host>

<port>9000</port>

</replica>

</shard>

</slicer>

</clickhouse_remote_servers>

<zookeeper-servers>

<node index="1">

<host>10.246.1.115</host>

<port>2181</port>

</node>

<node index="2">

<host>10.246.1.93</host>

<port>2181</port>

</node>

</zookeeper-servers>

</yandex>


I start with default zookeeper config and empty data directories on both servers:

service clickhouse-server stop

service zookeeper-server stop

rm -rf /var/lib/clickhouse/data/slicer0* /var/lib/clickhouse/metadata/slicer0*

rm -rf /var/lib/zookeeper/*

service zookeeper-server init

service zookeeper-server start

service clickhouse-server start


Init tables on both servers:

cat <<EOF | clickhouse-client --multiquery;

CREATE DATABASE slicer00;

CREATE DATABASE slicer01;


CREATE TABLE slicer00.tesd (

siteId UInt32,

date Date,

docUuid FixedString(37),

metricV UInt32,

metricS0 UInt32,

metricS1 UInt32,

metricS2 UInt32,

metricTS0 UInt32,

metricTS1 UInt32,

metricTS2 UInt32,

metricX UInt32

) ENGINE = ReplicatedMergeTree('/clickhouse/tables/slicer00/tesd', 'server0', date, (siteId, date, docUuid), 8192);


CREATE TABLE slicer01.tesd (

siteId UInt32,

date Date,

docUuid FixedString(37),

metricV UInt32,

metricS0 UInt32,

metricS1 UInt32,

metricS2 UInt32,

metricTS0 UInt32,

metricTS1 UInt32,

metricTS2 UInt32,

metricX UInt32

) ENGINE = ReplicatedMergeTree('/clickhouse/tables/slicer01/tesd', 'server0', date, (siteId, date, docUuid), 8192);


CREATE TABLE tesd_all AS tesd ENGINE = Distributed(slicer, '', tesd, cityHash64(docUuid));

EOF


Insert only one row in tesd_all:

cat <<EOF | clickhouse-client --query="INSERT INTO tesd_all FORMAT TabSeparated"

1 1970-01-02 F7336ADC-8E70-11E7-A061-A2AE6567C489 1 1 1 1 1 1 1 1

EOF


And finally select from tesd_all:

# clickhouse-client --query="select * from tesd_all"

1 1970-01-02 F7336ADC-8E70-11E7-A061-A2AE6567C489\0 1 1 1 1 1 1 1 1

1 1970-01-02 F7336ADC-8E70-11E7-A061-A2AE6567C489\0 1 1 1 1 1 1 1 1


Any ideas?



понедельник, 4 сентября 2017 г., 18:57:45 UTC+3 пользователь Kost Kost написал:
Hello!

Kost Kost

unread,
Sep 6, 2017, 8:07:35 AM9/6/17
to ClickHouse
oh sorry, google has destroyed all indentations in config (((


понедельник, 4 сентября 2017 г., 18:57:45 UTC+3 пользователь Kost Kost написал:
Hello!

Николай Кочетов

unread,
Sep 6, 2017, 9:08:18 AM9/6/17
to ClickHouse
The problem should be in <default_database> tag. It should be specified for each replica, which means, that tag should be inside

<remote_servers>
  <slicer> <!-- your cluster -->
    <shard>
      <replica>
        <default_database>slicer00</default_database> <!-- here -->
        ...
      </replica>

      <replica>
        <default_database>slicer00</default_database> <!-- again -->
        ...
      </replica>
    </shard>
      <replica>
        <default_database>slicer01</default_database> <!-- other database -->
        ...
      </replica>

      <replica>
        <default_database>slicer01</default_database> <!-- again -->
        ...
      </replica>
    </shard>
    ...

Example.

Sorry if previous comment was misleading.

среда, 6 сентября 2017 г., 15:04:15 UTC+3 пользователь Kost Kost написал:

Kost Kost

unread,
Sep 6, 2017, 1:21:25 PM9/6/17
to ClickHouse
Thank you!
That was my fault I didn't mentioned your first notice "in cluster configuration for each replica."

It sounds oddly but now all my requests over Distributed table return randomly different dataset or even empty!
I have discovered in logs that problem appears only if request goes to the same(!) nodes. Maybe it's just a coincidence:
2017.09.06 17:05:09.860930 [ 22 ] <Debug> executeQuery: (from [::ffff:10.246.1.93]:42160) SELECT docUuid, SUM(metricV) AS view, SUM(metricTS2) AS ts2, ts2 / view AS conversion FROM tesd  WHERE (siteId = 3) AND (date >= '1970-03-01') AND (date < '1970-06-01') GROUP BY docUuid ORDER BY view DESC LIMIT 20
2017.09.06 17:05:09.861135 [ 18 ] <Debug> executeQuery: (from [::ffff:10.246.1.93]:42158) SELECT docUuid, SUM(metricV) AS view, SUM(metricTS2) AS ts2, ts2 / view AS conversion FROM tesd  WHERE (siteId = 3) AND (date >= '1970-03-01') AND (date < '1970-06-01') GROUP BY docUuid ORDER BY view DESC LIMIT 20

And result is always consistent if second server was manually turned off before the request.
It seems some kind of network timeouts happens while request but error log is empty and debug log is over verbosed to discover problem at the first glance.


среда, 6 сентября 2017 г., 16:08:18 UTC+3 пользователь Николай Кочетов написал:

Николай Кочетов

unread,
Sep 7, 2017, 8:40:48 AM9/7/17
to ClickHouse
I've just noticed that used <default_database>shard_0</default_database> in the first comment and <default_database>slicer00</default_database> in the second.
Could you show your current config again to avoid misunderstanding?

среда, 6 сентября 2017 г., 20:21:25 UTC+3 пользователь Kost Kost написал:

Kost Kost

unread,
Sep 7, 2017, 9:31:02 AM9/7/17
to ClickHouse
Here is my current config https://pastebin.com/7cYJ3yuV

Now everything seems to work correctly:
- data is not duplicated
- on each server i have different dataset in tables slicer00.tesd and slicer01.tesd

I have correct number of rows from query select count(1) from tesd_all, but sometimes this request gives me a different results or even empty!
Exploring logs gave no result so far. My suggestion of this unstable result is that clickhouse falls on some network timeout when requesting to another shard.
I literally use default clickhouse and zookeeper settings, maybe some tuning required in case of replication.


четверг, 7 сентября 2017 г., 15:40:48 UTC+3 пользователь Николай Кочетов написал:

Николай Кочетов

unread,
Sep 7, 2017, 10:56:53 AM9/7/17
to ClickHouse
Config seems ok. Also you can:
1. Check that zookeeper works properly. For example, connect from one host to zookeeper with zkCli.sh and create temporary node (/tmp or something). Then connect form the other host and check that ls / shows tmp.
2. Try to use hostname instead of ip in config. Can remember that it solved similar problem for other person.

четверг, 7 сентября 2017 г., 16:31:02 UTC+3 пользователь Kost Kost написал:
Reply all
Reply to author
Forward
0 new messages