RPC-прокси

В данном разделе приведены примеры работы с YT через RPC-прокси для разных языков программирования: C++, Java, Python.

С++

Пример на Github

В примере реализована программа, через которую можно работать с динамическими таблицами в интерактивном режиме.

Формат вызова:

$ ./rpc_proxy_sample --config config.yson --user <user> --token $(cat ~/.yt/token) 2> /dev/null

Работа с Кипарисом:

list //sys
get //sys

Чтение данных средствами языка запросов (SelectRows):

select timestamp, host, rack, utc_time, data FROM [//home/dev/autorestart_nodes_copy] LIMIT 10

Запрос строк по ключу (LookupRows):

ulookup //home/dev/autorestart_nodes_copy timestamp;host;rack;utc_time;data <id=0>1486113922563016;<id=1>"s04-sas.cluster-name";<id=2>"SAS2.4.3-13" <id=0>1486113924172063;<id=1>"s04-sas.cluster-name";<id=2>"SAS2.4.3-13" <id=0>1486113992045484;<id=1>"s04-sas.cluster-name";<id=2>"SAS2.4.3-13" <id=0>1486113992591731;<id=1>"s04-sas.cluster-name";<id=2>"SAS2.4.3-13"  <id=0>1486113997734536;<id=1>"n4137-sas.cluster-name";<id=2>"SAS2.4.3-13"

Вставка строк (aka WriteRows):

upsert //home/dev/autorestart_nodes_copy timestamp;host;rack;utc_time;data <id=0>123;<id=1>"host123";<id=2>"rack123";<id=3>"utc_time1";<id=4>"data1" <id=0>567;<id=1>"host567";<id=2>"rack567";<id=3>"utc_time2";<id=4>"data2"

Удаление данных (DeleteRows):

delete //home/dev/autorestart_nodes_copy timestamp;host;rack <id=0>123;<id=1>"host123";<id=2>"rack123"

Пример работы с транзакциями:

if (ValidateSignature("delete", {"path", "columns", "..."}, tokens)) {
    auto path = tokens[1];
    TPrepareRows prepareRows(tokens);

    auto tx = Client_->StartTransaction(NTransactionClient::ETransactionType::Tablet).Get();
    if (!ValidateResult(tx)) return;

    tx.Value()->DeleteRows(path, prepareRows.NameTable, prepareRows.Rows);
    auto result = tx.Value()->Commit().Get();
    if (!ValidateResult(result)) return;

    Cout << "Committed" << Endl;
}

Java

Исходные коды представленных примеров находятся в GitHub.

Создание клиента:

   public static YtClient createYtClient(BusConnector connector, String user, String token)
    {
        return new YtClient(connector, "cluster-name", RpcCredentials(user, token));
    }

Реализация Select запроса:

package ru.yandex.yt.ytclient.examples;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import ru.yandex.yt.ytclient.tables.ColumnSchema;
import ru.yandex.yt.ytclient.wire.UnversionedRow;
import ru.yandex.yt.ytclient.wire.UnversionedRowset;
import ru.yandex.yt.ytclient.wire.UnversionedValue;
import ru.yandex.yt.ytclient.ytree.YTreeMapNode;

public class SelectRowsExample {
    private static final Logger logger = LoggerFactory.getLogger(SelectRowsExample.class);

    public static void main(String[] args) {
        ExamplesUtil.runExampleWithBalancing(client -> {
            long t0 = System.nanoTime();
            UnversionedRowset rowset = client.selectRows(
                    "timestamp, host, rack, utc_time, data FROM [//home/dev/autorestart_nodes_copy] LIMIT 10")
                    .join();
            long t1 = System.nanoTime();
            logger.info("Request time: {}", (t1 - t0) / 1000000.0);
            logger.info("Result schema:");
            for (ColumnSchema column : rowset.getSchema().getColumns()) {
                logger.info("    {}", column.getName());
            }
            for (UnversionedRow row : rowset.getRows()) {
                logger.info("Row:");
                for (UnversionedValue value : row.getValues()) {
                    logger.info("    value: {}", value);
                }
            }
            for (YTreeMapNode row : rowset.getYTreeRows()) {
                logger.info("Row: {}", row);
            }
        });
    }
}

Работа с Кипарисом:

package ru.yandex.yt.ytclient.examples;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class GetNodeExample {
    private static final Logger logger = LoggerFactory.getLogger(GetNodeExample.class);

    public static void main(String[] args) {
        ExamplesUtil.runExample(client -> {
            logger.info("Table dynamic: {}", client.getNode("//home/dev/autorestart_nodes_copy/@dynamic").join());
            logger.info("Table schema: {}", client.getNode("//home/dev/autorestart_nodes_copy/@schema").join());
        });
    }
}

Работа с Кипарисом:

package ru.yandex.yt.ytclient.examples;

import java.util.Random;

import joptsimple.OptionParser;
import joptsimple.OptionSet;
import joptsimple.OptionSpec;

import ru.yandex.yt.rpcproxy.ETransactionType;
import ru.yandex.yt.ytclient.misc.YtGuid;
import ru.yandex.yt.ytclient.proxy.ApiServiceTransaction;
import ru.yandex.yt.ytclient.proxy.ApiServiceTransactionOptions;
import ru.yandex.yt.ytclient.proxy.request.ObjectType;

public class CypressExample {

    public static void main(String[] args) {
        try {
            OptionParser parser = new OptionParser();

            OptionSpec<String> proxyOpt = parser.accepts("proxy", "proxy (see //sys/rpc_proxies)")
                    .withRequiredArg().ofType(String.class);

            OptionSet option = parser.parse(args);

            String [] hosts = null;

            if (option.hasArgument(proxyOpt)) {
                String line = option.valueOf(proxyOpt);
                hosts = line.split(",");
            } else {
                parser.printHelpOn(System.out);
                System.exit(1);
            }

            Random rnd = new Random();
            String host = hosts[rnd.nextInt(hosts.length)];

            ExamplesUtil.runExample(client -> {
                try {
                    ApiServiceTransactionOptions transactionOptions =
                            new ApiServiceTransactionOptions(ETransactionType.MASTER);

                    String node = "//tmp/test-node-cypress-example";
                    ApiServiceTransaction t = client.startTransaction(transactionOptions).get();


                    t.existsNode(node).thenAccept(result -> {
                        try {
                            if (result) {
                                t.removeNode(node);
                            }
                        } catch (Throwable e) {
                            throw new RuntimeException(e);
                        }
                    }).get();

                    YtGuid guid = t.createNode(node, ObjectType.Table).get();
                    /*
                    Map<String, YTreeNode> data = new HashMap<String, YTreeNode>();
                    data.put("k1", new YTreeInt64Node(10, new HashMap<>()));
                    data.put("k2", new YTreeInt64Node(31337, new HashMap<>()));
                    data.put("str", new YTreeStringNode("stroka"));
                    t.setNode(node, new YTreeMapNode(data)).get();
                    */
                    t.commit();

                    ApiServiceTransaction t2 = client.startTransaction(transactionOptions).get();
                    t2.linkNode(node, node + "-link");
                    t2.moveNode(node, node + "-moved");
                    t2.commit();

                    client.removeNode(node + "-link");

                } catch (Throwable e) {
                    System.out.println(e);
                    e.printStackTrace();
                    System.exit(-1);
                }
            }, ExamplesUtil.getUser(), ExamplesUtil.getToken(), host);

            System.exit(0);
        } catch (Throwable e) {
            System.out.println(e);
            e.printStackTrace();
            System.exit(-1);
        }
    }
}

Выполнение LookupRows:

package ru.yandex.yt.ytclient.examples;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import ru.yandex.yt.ytclient.proxy.LookupRowsRequest;
import ru.yandex.yt.ytclient.tables.ColumnSchema;
import ru.yandex.yt.ytclient.tables.ColumnValueType;
import ru.yandex.yt.ytclient.tables.TableSchema;
import ru.yandex.yt.ytclient.wire.UnversionedRow;
import ru.yandex.yt.ytclient.wire.UnversionedRowset;
import ru.yandex.yt.ytclient.wire.UnversionedValue;
import ru.yandex.yt.ytclient.ytree.YTreeMapNode;

public class LookupRowsExample {
    private static final Logger logger = LoggerFactory.getLogger(LookupRowsExample.class);

    public static void main(String[] args) {
        TableSchema schema = new TableSchema.Builder()
                .addKey("timestamp", ColumnValueType.INT64)
                .addKey("host", ColumnValueType.STRING)
                .addKey("rack", ColumnValueType.STRING)
                .addValue("utc_time", ColumnValueType.STRING)
                .addValue("data", ColumnValueType.STRING)
                .build();
        ExamplesUtil.runExampleWithBalancing(client -> {
            long t0 = System.nanoTime();
            LookupRowsRequest request = new LookupRowsRequest("//home/dev/autorestart_nodes_copy", schema.toLookup())
                    .addFilter(1486113922563016L, "s04-sas.cluster-name", "SAS2.4.3-13")
                    .addFilter(1486113924172063L, "s04-sas.cluster-name", "SAS2.4.3-13")
                    .addFilter(1486113992045484L, "s04-sas.cluster-name", "SAS2.4.3-13")
                    .addFilter(1486113992591731L, "s04-sas.cluster-name", "SAS2.4.3-13")
                    .addFilter(1486113997734536L, "n4137-sas.cluster-name", "SAS2.4.3-13")
                    .addLookupColumns("utc_time", "data");
            long t1 = System.nanoTime();
            UnversionedRowset rowset = client.lookupRows(request).join();
            long t2 = System.nanoTime();
            logger.info("Request time: {}ms + {}ms", (t1 - t0) / 1000000.0, (t2 - t1) / 1000000.0);
            logger.info("Result schema:");
            for (ColumnSchema column : rowset.getSchema().getColumns()) {
                logger.info("    {}", column.getName());
            }
            for (UnversionedRow row : rowset.getRows()) {
                logger.info("Row:");
                for (UnversionedValue value : row.getValues()) {
                    logger.info("    value: {}", value);
                }
            }
            for (YTreeMapNode row : rowset.getYTreeRows()) {
                logger.info("Row: {}", row);
            }
        });
    }
}

Выполнение ModifyRows:

package ru.yandex.yt.ytclient.examples;

import java.util.Arrays;
import java.util.UUID;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import ru.yandex.yt.rpcproxy.ETransactionType;
import ru.yandex.yt.ytclient.proxy.ApiServiceTransaction;
import ru.yandex.yt.ytclient.proxy.ApiServiceTransactionOptions;
import ru.yandex.yt.ytclient.proxy.LookupRowsRequest;
import ru.yandex.yt.ytclient.proxy.ModifyRowsRequest;
import ru.yandex.yt.ytclient.tables.ColumnSchema;
import ru.yandex.yt.ytclient.tables.ColumnValueType;
import ru.yandex.yt.ytclient.tables.TableSchema;
import ru.yandex.yt.ytclient.wire.UnversionedRow;
import ru.yandex.yt.ytclient.wire.UnversionedRowset;
import ru.yandex.yt.ytclient.wire.UnversionedValue;
import ru.yandex.yt.ytclient.ytree.YTreeBuilder;
import ru.yandex.yt.ytclient.ytree.YTreeMapNode;

public class ModifyRowsExample {
    private static final Logger logger = LoggerFactory.getLogger(ModifyRowsExample.class);

    public static void main(String[] args) {
        TableSchema schema = new TableSchema.Builder()
                .addKey("timestamp", ColumnValueType.INT64)
                .addKey("host", ColumnValueType.STRING)
                .addKey("rack", ColumnValueType.STRING)
                .addValue("utc_time", ColumnValueType.STRING)
                .addValue("data", ColumnValueType.STRING)
                .build();
        ExamplesUtil.runExample(client -> {
            ApiServiceTransactionOptions transactionOptions =
                    new ApiServiceTransactionOptions(ETransactionType.MASTER);
            try (ApiServiceTransaction transaction = client.startTransaction(transactionOptions).join()) {
                logger.info("Transaction started: {} (timestamp={}, ping={})",
                        transaction.getId(),
                        transaction.getStartTimestamp(),
                        transaction.isPing());

                transaction.ping().join();
                logger.info("Transaction ping succeeded!");

                ModifyRowsRequest request =
                        new ModifyRowsRequest("//home/dev/autorestart_nodes_copy", schema)
                                .addInsert(Arrays.asList(10, "myhost1", "myrack1", "utc_time1", "data1"))
                                .addInsert(Arrays.asList(11, "myhost2", "myrack2", "utc_time2", "data2"))
                                .addUpdate(new YTreeBuilder()
                                        .beginMap()
                                        .key("timestamp").value(1486190036109192L)
                                        .key("host").value("n0344-sas.cluster-name")
                                        .key("rack").value("SAS2.4.3-15")
                                        .key("data").value("XXX " + UUID.randomUUID().toString())
                                        .buildMap()
                                        .mapValue())
                                .addUpdate(new YTreeBuilder()
                                        .beginMap()
                                        .key("timestamp").value(1486190037953802L)
                                        .key("host").value("s03-sas.cluster-name")
                                        .key("rack").value("SAS2.4.3-15")
                                        .key("data").value("XXX " + UUID.randomUUID().toString())
                                        .buildMap()
                                        .mapValue());
                long t0 = System.nanoTime();
                transaction.modifyRows(request).join();
                long t1 = System.nanoTime();

                logger.info("Request time: {}ms", (t1 - t0) / 1000000.0);

                t0 = System.nanoTime();

                LookupRowsRequest lookup = new LookupRowsRequest("//home/dev/autorestart_nodes_copy", schema.toLookup())
                        .addFilter(1486190036109192L, "n0344-sas.cluster-name", "SAS2.4.3-15")
                        .addFilter(1486190037953802L, "s03-sas.cluster-name", "SAS2.4.3-15")
                        .addLookupColumns("timestamp", "data");


                t1 = System.nanoTime();
                UnversionedRowset rowset = client.lookupRows(lookup).join();
                long t2 = System.nanoTime();
                logger.info("Request time: {}ms + {}ms", (t1 - t0) / 1000000.0, (t2 - t1) / 1000000.0);
                logger.info("Result schema:");
                for (ColumnSchema column : rowset.getSchema().getColumns()) {
                    logger.info("    {}", column.getName());
                }
                for (UnversionedRow row : rowset.getRows()) {
                    logger.info("Row:");
                    for (UnversionedValue value : row.getValues()) {
                        logger.info("    value: {}", value);
                    }
                }
                for (YTreeMapNode row : rowset.getYTreeRows()) {
                    logger.info("Row: {}", row);
                }

                transaction.commit().join();
                logger.info("Transaction committed!");
            }
        });
    }
}

Python

Для работы через Python необходимо установить пакет с биндингами, pypi-пакет называется ytsaurus-rpc-driver. После этого необходимо указать в качестве бэкенда rpc.

Работа с динамическими таблицами:

import yt.wrapper as yt

def main():
    client = yt.YtClient("cluster-name", config={"backend": "rpc"})
    schema = [
        {"name": "x", "type": "int64", "sort_order": "ascending"},
        {"name": "y", "type": "int64"},
        {"name": "z", "type": "int64"}
    ]

    table = "//home/ignat/dynamic_table"
    client.create("table", table, attributes={"schema": schema, "dynamic": True})
    client.mount_table(table, sync=True)
    client.insert_rows(table, [{"x": 0, "y": 99}])

    for iter in xrange(5):
        with client.Transaction(type="tablet"):
            rows = list(client.lookup_rows(table, [{"x": 0}]))
            if len(rows) == 1 and rows[0]["y"] <= 100:
                rows[0]["y"] += 1
                client.insert_rows(table, rows)

    print list(client.select_rows("* from [{}]".format(table)))

if __name__ == "__main__":
    main()

Для того чтобы клиент ходил к проксям определённой роли (например, my_role), нужно указать в конфиге опцию proxy_role:

client = yt.YtClient("cluster-name", config={"backend": "rpc", "driver_config": {"proxy_role": "my_role"}})

Мы также советуем включать дебажные логи клиентской библиотеки. Такие логи сильно помогут при возникновении проблем. Помимо обычного способа через переменную окружения YT_LOG_LEVEL=debug можно явно настроить логирование внутри rpc proxy client. Для этого нужно передать конфиг логирования через параметр driver_logging_config:

logging_config = {
    'rules': [{'min_level': 'debug', 'writers': ['file']}],
    'writers': {'file': {'file_name': logging_file_name, 'type': 'file'}}
}
client = yt.YtClient("cluster-name", config={"backend": "rpc", "driver_logging_config": logging_config})
Предыдущая
HTTP-прокси
Следующая
Справочник по HTTP-прокси
В этой статье: