Data Replication and Configuration File

[alexandre...@gmail.com]

Hello,
I am working about replication tables (ReplicatedMergeTree) in three shards :

.<?xml version="1.0"?>
<yandex>

<!-- Configuration of clusters that could be used in Distributed tables. -->
    <remote_servers>
        <clickHouseCluster>
            <shard>
                <replica>
                    <host>clickhouse</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>clickhouse1</host>
                    <port>9000</port>
                </replica>
            </shard>
            <shard>
                <replica>
                    <host>clickhouse1</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>clickhouse2</host>
                    <port>9000</port>
                </replica>
            </shard>
            <shard>
                <internal_replication>true</internal_replication>
                <replica>
                    <host>clickhouse2</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>clickhouse</host>
                    <port>9000</port>
                </replica>
            </shard>
        </clickHouseCluster>
    </remote_servers>

    <zookeeper>
        <node>
            <host>kafka</host>
            <port>2181</port>
        </node>
    </zookeeper>

    <macros>
        <shard>01</shard>
        <replica>clickhouse1</replica>
    </macros>

    <macros>
        <shard>02</shard>
        <replica>clickhouse2</replica>
    </macros>

I have moreover a Distributed table on the server 1 (clickhouse), that chooses the server where the data are inserted from the sharding key.

I suceed in inserting data from the Distributed table to the Replicated tables.
The sum total of line for the three Replicated table is equal to 8800 (4400 insertions and 1 replica for each insertion).
But when I request the Distributed table with this request :

SELECT COUNT(*) FROM "my Distributed table"

I obtained two results : 7700 and 8069 lines.
I do not understand why the total of lines is different to 8800 and why it changes if I repeat the request.
Is it a normal behaviour of ClickHouse ?

I try to test the internal replication option.

<internal_replication>true</internal_replication>

I try for each of my shard separately but :
- for the first shard, it does not change these results
- for the second shard, the number of lines for the three servers are not equal to 8800 (it is equal to 6732) and there is always the problem of total of lines for the Distributed table.

So even though this option is recommended for Distributed table pointed at Replicated table, it clearly does not work.
Do you know how to solve my problem ?

Thank you for your help

[ste...@activitystream.com]

Hi Alexandre,

I'm new here and almost without a clue :).

I would think that a single instance of CH (machine or otherwise) should only be a part of one shard.
For me it would make sense that data is distributed at save time and not checked for duplication on read time.

That could mean, with your config,  that your data is duplicated because the machines carry more than one copy of some messages (based on the sharding).

Anyways; not sure that helps at all but please share your findings.

Regards,

 -Stefán

[alexandre...@gmail.com]

Hello Stefán,

Thank you for your response

If I keep just one instance of ClickHouse on each shard, there is no replication (if I have understand this concept).
In fact, this configuration allows me to have data in two shards.

Maybe I have problems when I create Replicated tables. I create my tables like this :

CREATE TABLE IF NOT EXISTS mtm_trade(job_id String, date Date, scenario Int64, trade String, mtm Float64)
      ENGINE =
        ReplicatedMergeTree(
          '/clickhouse/tables/02/mtm_trade',
          'clickhouse1',
          date,
          job_id,
          8192
        )
CREATE TABLE IF NOT EXISTS mtm_trade(job_id String, date Date, scenario Int64, trade String, mtm Float64)
      ENGINE =
        ReplicatedMergeTree(
          '/clickhouse/tables/03/mtm_trade',
          'clickhouse2',
          date,
          job_id,
          8192
        )

CREATE TABLE IF NOT EXISTS mtm_trade(job_id String, date Date, scenario Int64, trade String, mtm Float64)
      ENGINE =
        ReplicatedMergeTree(
          '/clickhouse/tables/01/mtm_trade',
          'clickhouse',
          date,
          job_id,
          8192
        )

 
I am not sure if I use correctly the good method to declare hits and replica.

ReplicatedMergeTree('/clickhouse/tables/{layer}-{shard}/hits', '{replica}', EventDate, intHash32(UserID), ..., 8192

Does anyone have suceed in both load balancing and data replication ?

Regards,
Alexandre

[ste...@activitystream.com]

The replication should come from having more machines, I think, if you are set on sharding by 3.

Especially if I understand this part of the documentation correctly: "When specifying replicas, one of the available replicas will be selected for reading"

Regards,

 -Stefán

[alexandre...@gmail.com]

To my mind, the replication allows me to have each data inserted in database two times or more to limit the risk of data loss if a server shuts down.
Though data must be replicated in other servers of my system.

Here I just try to apply this to a 3 three servers with a replication factor of 2.
After suceeding in this test, I will implement this on much more server with a replication factor of 3.
But I need to find the good results on a smaller example before implementing this.

I have the feeling that the replication of the server where the Distributed table is, caused problems when I select data after.

Regards,
Alexandre

[alexandre...@gmail.com]

Nobody knows how to implement replication and what are the steps to suceed in this task ?
(How configure the .xml file, how create the different tables needed ?)

Moreover I do not really understand the concept of macro. In fact, we can create Replicated Tables without macro.
Does anynone have further information ?

[tatiana....@revjet.com]

You don't have to use macros. They are convenient when you want to use the same DDL statement on different servers, but you don't have to use them.

To replicate data through ZooKeeper you need to create two tables with the same path in ZooKeeper, but different replica names. If you do that, and insert data in either one of the two tables, the data will show up in the second table as well.

[Alex Zatelepin]

Hi!

What you are trying to achieve is called cross-replication. I.e. you want 3 shards with 2 replicas each on a 3-machine cluster. It is possible, although it is less convenient and the recommended approach is to setup a 3х2=6-machine cluster.

The trick is to store tables that reside on the same server but are related to different shards in the different databases. You can then instruct Distributed tables to substitute correct database during query rewriting.

Here is an in-depth explanation (in Russian): https://groups.google.com/forum/#!topic/clickhouse/GFl04GsTNZg

Here is a configuration example: https://github.com/yandex/ClickHouse/issues/437

Macros are just a convenient feature allowing setting up replicas with the same statement on different machines (the difference is encapsulated into the config files). Using them is not required.

[alexandre...@gmail.com]

Thank for your response Tatiana and Alex !

I suceed in creating my cross-replication system with the good insertion in each server.
But how data are recovering if a server shuts down ?

According to the code of the configuration example :

Server 1.

default.graphite '/clickhouse_perftest/tables/01/graphite', 'r1'

replica.graphite '/clickhouse_perftest/tables/03/graphite', 'r2'

Server 2.

default.graphite '/clickhouse_perftest/tables/02/graphite', 'r1'

replica.graphite '/clickhouse_perftest/tables/01/graphite', 'r2'

Server 3.

default.graphite '/clickhouse_perftest/tables/03/graphite', 'r1'

replica.graphite '/clickhouse_perftest/tables/02/graphite', 'r2

If the server 2 shuts down, and, if I request the distributed table on the server 1 with this request :

SELECT COUNT(*) FROM "Distributed table"

I have the following response :

Code: 279. DB::Exception: Received from localhost:35196, 127.0.0.1. DB::NetException. DB::NetException: All connection tries failed. Log:

Code: 32, e.displayText() = DB::Exception: Attempt to read after eof, e.what() = DB::Exception
Code: 209, e.displayText() = DB::NetException: Timeout: connect timed out: 172.18.0.2:9000: (clickhouse1:9000, 172.18.0.2), e.what() = DB::NetException
Code: 209, e.displayText() = DB::NetException: Timeout: connect timed out: 172.18.0.2:9000: (clickhouse1:9000, 172.18.0.2), e.what() = DB::NetException

Normally, with this system, I must recover data of server 2 from his replica on the server 3, no ?

According to the documentation, I have to recover data manually before requesting my database. Is it right ?

[ste...@activitystream.com]

Hi Alexander,

Can you share the config files and the last part of your create tables part?

Regards,

 -Stefán

[alexandre...@gmail.com]

My configuration file is the following one :

<?xml version="1.0"?>
<yandex>

<!-- Configuration of clusters that could be used in Distributed tables. -->
    <remote_servers>
        <clickHouseCluster>
            <shard>

                <internal_replication>true</internal_replication>
                <replica>

                    <host>clickhouse</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>clickhouse1</host>
                    <port>9000</port>
                </replica>
            </shard>
            <shard>

</yandex>

On the server clickhouse, I create the tables :

CREATE TABLE IF NOT EXISTS mtm_trade(job_id String, date Date, scenario Int64, trade String, mtm Float64)
      ENGINE =
          ReplicatedMergeTree(
              '/clickhouse/tables/01/mtm_trade',

              'r1',
              date,
              ((job_id, trade), (job_id, scenario, date)),
              8192
          )

CREATE TABLE IF NOT EXISTS replica.mtm_trade(job_id String, date Date, scenario Int64, trade String, mtm Float64)

      ENGINE =
        ReplicatedMergeTree(
          '/clickhouse/tables/02/mtm_trade',

          'r2',
          date,
          ((job_id, trade), (job_id, scenario, date)),
          8192
        )

On the server 2, I create the tables :

CREATE TABLE IF NOT EXISTS mtm_trade(job_id String, date Date, scenario Int64, trade String, mtm Float64)
      ENGINE =
        ReplicatedMergeTree(
          '/clickhouse/tables/02/mtm_trade',

          'r1',
          date,
          ((job_id, trade), (job_id, scenario, date)),
          8192
        )

CREATE TABLE IF NOT EXISTS replica.mtm_trade(job_id String, date Date, scenario Int64, trade String, mtm Float64)

      ENGINE =
        ReplicatedMergeTree(
          '/clickhouse/tables/03/mtm_trade',

          'r2',
          date,
          ((job_id, trade), (job_id, scenario, date)),
          8192
        )

On the server 3, I create the tables :

CREATE TABLE IF NOT EXISTS mtm_trade(job_id String, date Date, scenario Int64, trade String, mtm Float64)
      ENGINE =
        ReplicatedMergeTree(
          '/clickhouse/tables/03/mtm_trade',

          'r1',
          date,
          ((job_id, trade), (job_id, scenario, date)),
          8192
        )

CREATE TABLE IF NOT EXISTS replica.mtm_trade(job_id String, date Date, scenario Int64, trade String, mtm Float64)

      ENGINE =
        ReplicatedMergeTree(
          '/clickhouse/tables/01/mtm_trade',

          'r2',
          date,
          ((job_id, trade), (job_id, scenario, date)),
          8192
        )

Finally I create the distributed table on the server 1 :

CREATE TABLE IF NOT EXISTS mtm_trade_global AS mtm_trade
      ENGINE =
          Distributed(
              clickHouseCluster,
              default,
              mtm_trade,
            intHash64(scenario)
          )

[ste...@activitystream.com]

Great, thank you

[Alex Zatelepin]

So for cross-replication to work you should configure default_database for each replica in your cluster config and set database to empty string when creating Distributed table.

[alexandre...@gmail.com]

Thank for your response.

The cross-replication works but if a server shuts down, I can not access and so read the data from this server.
Though I lost data and the replication is useless.
It's problematic, is there any solution ?

[Alex Zatelepin]

if a server shuts down, I can not access and so read the data from this server.

That's not how it is supposed to work. 3-server setup should tolerate the loss of any one server.

To validate that everything works as expected and for the future reference I've created a test: https://github.com/yandex/ClickHouse/commit/c90ea346ce9ef51886875b8f3f99c24c6c19c2e0

Please take a look, you can find configuration example there.

[alexandre...@gmail.com]

Thank you a lot for your help.

It effectively works. I make a confusion between shard and ClickHouse instance when I created Replicated tables.

[alexandre...@gmail.com]

Hello,

I implement this solution on my company's data centers after succeeding implementing this on Docker clusters.
However, data are not replicated ..

I use the same scripts to make the configuration file and to create tables on ClickHouse. (I verify the results and it seems to be great.)
I use the same version of Zookeeper 3.4.10 (but I use 3 Zookeeper instances instead of 1).

Data are correctly divided between servers, all my data are inserted in ClickHouse but data are only present once.

Do you any idea for my problem ?

[alexandre...@gmail.com]

It was a DNS error between servers.
I use IP address to define the servers and as ClickHouse is used in Docker on my servers, I have to add in /etc/hosts in Docker the different IP adress of my servers.

And it finally works.