Usage examples

You need to get a YTsaurus token before running the examples.

Run the examples on Linux, with no arguments.

We recommend setting the log level to INFO or DEBUG to make programs output links to the web interface where you can monitor the execution.

The log level is set using an environment variable with this command (more info is available in the documentation):

export YT_LOG_LEVEL=INFO

For production processes, we recommend saving the most detailed DEBUG logs. If any problem occurs, these logs will help you find out what was happening.

Basic level

Simple Map (TNode version)

Suppose there is a table with logins. You want to make a table with email addresses, deriving them from the login: email = login + "@domain".

To do this, you can use a simple mapper.

The example is located at yt/cpp/mapreduce/examples/tutorial/simple_map_tnode.

 'yt/cpp/mapreduce/tutorial/simple_map_tnode/main.cpp'

Simple Map (Protobuf version)

If you know the structure of the table, you may use the Protobuf format, which:

  • Is a bit faster.
  • Prevents such errors as typos in column names or usage of wrong type for column.

The example is located at yt/cpp/mapreduce/examples/tutorial/simple_map_protobuf.

import "yt/yt_proto/yt/formats/extension.proto";

message TLoginRecord
{
    optional string Name = 1 [(NYT.column_name) = "name"];
    optional string Login = 2 [(NYT.column_name) = "login"];
}

message TEmailRecord
{
    optional string Name = 1 [(NYT.column_name) = "name"];
    optional string Email = 2 [(NYT.column_name) = "email"];
}

#include <yt/cpp/mapreduce/examples/tutorial/simple_map_protobuf/data.pb.h>

#include <yt/cpp/mapreduce/interface/client.h>
#include <yt/cpp/mapreduce/interface/config.h>

#include <util/stream/output.h>
#include <util/system/user.h>

using namespace NYT;

class TComputeEmailsMapper
    : public IMapper<TTableReader<TLoginRecord>, TTableWriter<TEmailRecord>>
{
public:
    void Do(TReader* reader, TWriter* writer) override
    {
        for (auto& cursor : *reader) {
            const auto& loginRecord = cursor.GetRow();

            TEmailRecord emailRecord;
            emailRecord.set_name(loginRecord.name());
            emailRecord.set_email(loginRecord.login() + "@yandex-team.ru");

            writer->AddRow(emailRecord);
        }
    }
};
REGISTER_MAPPER(TComputeEmailsMapper)

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");

    // Выходная табличка у нас будет лежать в tmp и содержать имя текущего пользователя.
    const TString outputTable = "//tmp/" + GetUsername() + "-tutorial-emails-protobuf";

    client->Map(
        TMapOperationSpec()
            .AddInput<TLoginRecord>("//home/tutorial/staff_unsorted")
            .AddOutput<TEmailRecord>(outputTable),
        new TComputeEmailsMapper);

    Cout << "Output table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << outputTable << Endl;

    return 0;
}

Simple Map with Protobuf using lambda expressions in C++

Some simple operations can be represented as C++ lambda expressions with an empty capture-list (not capturing variables) using the yt/cpp/mapreduce/library/lambda library.

The previous example can be rewritten as follows (data.proto is the same):

(yt/cpp/mapreduce/examples/tutorial/simple_map_lambda)

#include <yt/cpp/mapreduce/examples/tutorial/simple_map_lambda/data.pb.h>

#include <yt/cpp/mapreduce/library/lambda/yt_lambda.h>
#include <yt/cpp/mapreduce/interface/config.h>

#include <util/stream/output.h>
#include <util/system/user.h>

struct TGlobalSettings {
    TString MailSuffix;
    Y_SAVELOAD_DEFINE(MailSuffix);
};
NYT::TSaveable<TGlobalSettings> GlobalSettings;

int main() {
    NYT::Initialize();

    auto client = NYT::CreateClient("freud");

    // Выходная табличка у нас будет лежать в tmp и содержать имя текущего пользователя.
    auto outputTable = NYT::TRichYPath("//tmp/" + GetUsername() + "-tutorial-emails-protobuf")
        .CompressionCodec("zlib_9");

    // TransformCopyIf запоминает содержимое GlobalSettings на момент своего запуска
    // и восстанавливает его в job'ах, в которых вызывается лямбда.
    // Этот способ требует аккуратности, если операции запускать из разных потоков.
    GlobalSettings.MailSuffix = "@yandex-team.ru";

    NYT::TransformCopyIf<TLoginRecord, TEmailRecord>(
        client,
        "//home/tutorial/staff_unsorted",
        outputTable,
        [](auto& src, auto& dst) {
            dst.set_name(src.name());
            dst.set_email(src.login() + GlobalSettings.MailSuffix);
            return true;
        });

    Cout << "Output table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << outputTable.Path_ << Endl;

    return 0;
}

Sorting a table and a simple Reduce operation

Given the same input table, we can count how many times each name occurs. To do this we need Reduce operation. Reduce operation requires input table to be sorted, so we run the Sort operation before launching Reduce.

The example is located at yt/cpp/mapreduce/examples/tutorial/simple_reduce_tnode.

#include <yt/cpp/mapreduce/interface/client.h>

#include <util/stream/output.h>
#include <util/system/user.h>

using namespace NYT;

class TCountNamesReduce
    : public IReducer<TTableReader<TNode>, TTableWriter<TNode>>
{
public:
    void Do(TReader* reader, TWriter* writer) override
    {
        // В Do приходят все записи с общим reduce ключом, т.е. в нашем случае с общим полем `name'.
        const auto& row = reader->GetRow();
        const auto name = row["name"];

        ui32 count = 0;
        for ([[maybe_unused]] auto& cursor : *reader) {
            ++count;
        }

        TNode result;
        result["name"] = name;
        result["count"] = count;
        writer->AddRow(result);
    }
};
REGISTER_REDUCER(TCountNamesReduce)

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");

    const TString sortedTmpTable = "//tmp/" + GetUsername() + "-tutorial-tmp";
    const TString outputTable = "//tmp/" + GetUsername() + "-tutorial-emails";

    client->Sort(
        TSortOperationSpec()
            .AddInput("//home/tutorial/staff_unsorted")
            .Output(sortedTmpTable)
            .SortBy({"name"}));

    client->Reduce(
        TReduceOperationSpec()
            .ReduceBy({"name"})
            .AddInput<TNode>(sortedTmpTable) // Входная таблица должна быть отсортирована по тем колонкам,
                                             // по которым мы осуществляем reduce.
            .AddOutput<TNode>(outputTable),
        new TCountNamesReduce);

    Cout << "Output table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << outputTable << Endl;

    return 0;
}

Composite types and Protobuf format (using Reduce as an example)

Let's say we have a table with a set of records on links occurring in web documents. This table has the following columns:

  • DocTitle: is a string containing the document name.
  • Link is a struct containing following fields:
    • Host: host part of the link url.
    • Port: port part of the link url.
    • Path: path part of the link url.
  • OccurrenceCount: how many times link appears in the document.

Our input table is sorted by DocTitle column. We want to aggregate information about each document into single row of output table.

Let's describe protobuf messages of the following form:

import "yt/yt_proto/yt/formats/extension.proto";

message TUrl {
    optional string host = 1 [(NYT.column_name) = "Host"];
    optional int32 port = 2 [(NYT.column_name) = "Port"];
    optional string path = 3 [(NYT.column_name) = "Path"];
}

message TExtraInfo {
    optional uint32 total_occurrence_count = 1;
}

message TDoc {
    // Помечаем, что поля данного сообщения будут сериализоваться
    // в YT-режиме. Это означает, что им сопоставлены колонки таблицы
    // с соответствующими (возможно, сложными) типами.
    option (NYT.default_field_flags) = SERIALIZATION_YT;

    optional string title = 1            [(NYT.key_column_name) = "Title"];
    repeated TUrl links = 2              [(NYT.column_name) = "Links"];
    repeated uint32 occurrence_count = 3 [(NYT.column_name) = "UpdateTimes"];

    // Данное поле специально помечено как сериализуемое в бинарном (PROTOBUF) режиме.
    // В таблице эта колонка будет иметь тип "string".
    optional TExtraInfo extra_info = 4    [(NYT.column_name) = "ExtraInfo", (NYT.flags) = SERIALIZATION_PROTOBUF];
}

message TLinkEntry {
    option (NYT.default_field_flags) = SERIALIZATION_YT;
    optional string doc_title = 1        [(NYT.column_name) = "DocTitle"];
    optional TUrl link = 2              [(NYT.column_name) = "Link"];
    optional uint32 occurrence_count = 3 [(NYT.column_name) = "OccurrenceCount"];
}

TLinkEntry corresponds to the input table row, and TDoc corresponds to the output table row.

The TDoc fields have the following semantics:

  • Title: Title.
  • Links: List of links occurring in the document.
  • OccurrenceCounts: List with the number of times the given links occur (this list has same length as the Links list).
  • ExtraInfo: Additional information, which in this case includes the total number of times the links occurs.

Attention

(NYT.field_serialization_mode) in TDoc and TLinkEntry messages. By default, this option equals PROTOBUF, which means "serialize the field as a sequence of bytes". The YT value we set means that the corresponding nested message will be related to the composite type in the table schema. In TDoc, as an example, we marked the ExtraInfo field with the (NYT.serialization_mode) = PROTOBUF option (note the type of the respective field in the output table).

The reducer is straightforward.

#include <yt/cpp/mapreduce/examples/tutorial/protobuf_complex_types/data.pb.h>

#include <yt/cpp/mapreduce/interface/client.h>
#include <yt/cpp/mapreduce/interface/config.h>

#include <util/stream/output.h>
#include <util/system/user.h>

using namespace NYT;

// Редьюсер агрегирует информацию о ссылках на документ с данным заголовком.
class AggregateLinksReducer
    : public IReducer<TTableReader<TLinkEntry>, TTableWriter<TDoc>>
{
public:
    void Do(TReader* reader, TWriter* writer) override
    {
        TDoc doc;
        for (auto& cursor : *reader) {
            auto entry = cursor.MoveRow();
            if (!doc.has_title()) {
                doc.set_title(entry.doc_title());
            }
            doc.add_links()->Swap(entry.mutable_link());
            doc.add_occurrence_count(entry.occurrence_count());
            auto newCount = doc.extra_info().total_occurrence_count() + entry.occurrence_count();
            doc.mutable_extra_info()->set_total_occurrence_count(newCount);
        }
        writer->AddRow(doc);
    }
};
REGISTER_REDUCER(AggregateLinksReducer)

int main() {
    NYT::Initialize();

    TString cluster = "freud";
    auto client = CreateClient(cluster);

    const TString sortedLinksTable  = "//home/tutorial/links_sorted_schematized";

    Cout << "Sorted links table: https://yt.yandex-team.ru/" << cluster << "/#page=navigation&offsetMode=row&path=" << sortedLinksTable << Endl;

    const TString outputTable = "//tmp/" + GetUsername() + "-tutorial-docs-protobuf";

    // Обратите внимание на опцию `InferOutputSchema`,
    // она заставляет навешивать на выходную таблицу схему.
    client->Reduce(
        TReduceOperationSpec()
            .AddInput<TLinkEntry>(sortedLinksTable)
            .AddOutput<TDoc>(outputTable)
            .ReduceBy({"DocTitle"}),
        new AggregateLinksReducer,
        TOperationOptions()
            .InferOutputSchema(true));

    Cout << "Output table: https://yt.yandex-team.ru/" << cluster << "/#page=navigation&offsetMode=row&path=" << outputTable << Endl;

    return 0;
}

Reduce with multiple input tables

Imagine that besides the table with users from our previous examples, there is also a second table with robots.

The following program creates a table where non-robot users are filtered out.

The example is located at yt/cpp/mapreduce/examples/tutorial/multiple_input_reduce_tnode.

#include <yt/cpp/mapreduce/interface/client.h>

#include <util/stream/output.h>
#include <util/system/user.h>

using namespace NYT;

class TFilterRobotsReduce
    : public IReducer<TTableReader<TNode>, TTableWriter<TNode>>
{
public:
    void Do(TReader* reader, TWriter* writer) override {
        TNode loginRow;
        bool isRobot = false;
        for (auto& cursor : *reader) {
            const auto& curRow = cursor.GetRow();

            // У нас есть информация о том из какой таблицы пришла запись.
            auto tableIndex = cursor.GetTableIndex();
            if (tableIndex == 0) {
                // Таблица с логинами.
                loginRow = curRow;
            } else if (tableIndex == 1) {
                // Таблица про роботов.
                isRobot = curRow["is_robot"].AsBool();
            } else {
                // Какая-то фигня, такого индекса быть не может.
                Y_ABORT();
            }
        }

        if (isRobot) {
            writer->AddRow(loginRow);
        }
    }
};
REGISTER_REDUCER(TFilterRobotsReduce)

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");

    const TString sortedLoginTable = "//tmp/" + GetUsername() + "-tutorial-login-sorted";
    const TString sortedIsRobotTable = "//tmp/" + GetUsername() + "-tutorial-is_robot-sorted";
    const TString outputTable = "//tmp/" + GetUsername() + "-tutorial-robots";

    client->Sort(
        TSortOperationSpec()
            .AddInput("//home/tutorial/staff_unsorted")
            .Output(sortedLoginTable)
            .SortBy({"uid"}));

    client->Sort(
        TSortOperationSpec()
            .AddInput("//home/tutorial/is_robot_unsorted")
            .Output(sortedIsRobotTable)
            .SortBy({"uid"}));

    client->Reduce(
        TReduceOperationSpec()
            .ReduceBy({"uid"})
            .AddInput<TNode>(sortedLoginTable) // Таблицу с логинами мы добавляем первой, поэтому её TableIndex == 0
            .AddInput<TNode>(sortedIsRobotTable) // Таблицу про роботов мы добавляем второй, поэтому её TableIndex == 1
            .AddOutput<TNode>(outputTable),
        new TFilterRobotsReduce);

    Cout << "Output table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << outputTable << Endl;
    return 0;
}

Reading files from operations

The previous task can be solved in a different way: upload the table with robots directly to the machines running the operations and read it there as a whole in yhash_set.

In this case, you don't have to sort the table and can use Map instead of Reduce. This method is used for small tables up to several GB in size.

Input tables are the same as in the previous example: with users and with robots.

The following program creates a table where non-robot users are filtered out.

The example is located at yt/cpp/mapreduce/examples/tutorial/map_tnode_with_file.

#include <yt/cpp/mapreduce/interface/client.h>
#include <yt/cpp/mapreduce/interface/io.h>

#include <util/generic/hash_set.h>
#include <util/stream/output.h>
#include <util/system/user.h>

using namespace NYT;

class TFilterRobotsMap
    : public IMapper<TTableReader<TNode>, TTableWriter<TNode>>
{
public:
    void Do(TReader* loginReader, TWriter* writer) override {
        // Если мы прикрепляли табличку //path/to/table, то в операции мы будем видеть её под именем table
        TIFStream stream("is_robot_unsorted");
        auto isRobotReader = CreateTableReader<TNode>(&stream);
        THashSet<i64> robotIds;
        for (auto& cursor : *isRobotReader) {
            const auto& curRow = cursor.GetRow();
            if (curRow["is_robot"].AsBool()) {
                robotIds.insert(curRow["uid"].AsInt64());
            }
        }

        for (auto& cursor : *loginReader) {
            const auto& curRow = cursor.GetRow();
            if (robotIds.contains(curRow["uid"].AsInt64())) {
                writer->AddRow(curRow);
            }
        }
    }
};
REGISTER_MAPPER(TFilterRobotsMap)

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");


    const TString loginTable = "//home/tutorial/staff_unsorted";
    const TString isRobotTable = "//home/tutorial/is_robot_unsorted";
    const TString outputTable = "//tmp/" + GetUsername() + "-tutorial-robots";

    client->Map(
        TMapOperationSpec()
            .AddInput<TNode>(loginTable)
            .MapperSpec(TUserJobSpec()
                .AddFile(TRichYPath(isRobotTable).Format("yson"))) // Таблицу с роботами добавляем в виде файла
            .AddOutput<TNode>(outputTable),
        new TFilterRobotsMap);

    Cout << "Output table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << outputTable << Endl;
    return 0;
}

Reduce with multiple input and output tables

Here we have a similar task as in the previous example, except this time we'll write two output tables at once: with human users and with robots.

The example is located at yt/cpp/mapreduce/examples/tutorial/multiple_input_multiple_output_reduce_tnode.

#include <yt/cpp/mapreduce/interface/client.h>

#include <util/stream/output.h>
#include <util/system/user.h>

using namespace NYT;

class TSplitHumanRobotsReduce
    : public IReducer<TTableReader<TNode>, TTableWriter<TNode>>
{
public:
    void Do(TReader* reader, TWriter* writer) override {
        TNode loginRow;
        bool isRobot = false;
        for (auto& cursor : *reader) {
            const auto& curRow = cursor.GetRow();

            auto tableIndex = cursor.GetTableIndex();
            if (tableIndex == 0) {
                loginRow = curRow;
            } else if (tableIndex == 1) {
                isRobot = curRow["is_robot"].AsBool();
            } else {
                Y_ABORT();
            }
        }

        // Второй аргумент метода `AddRow' указывает, в какую таблицу будет записано значение.
        if (isRobot) {
            writer->AddRow(loginRow, 0);
        } else {
            writer->AddRow(loginRow, 1);
        }
    }
};
REGISTER_REDUCER(TSplitHumanRobotsReduce)

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");

    const TString sortedLoginTable = "//tmp/" + GetUsername() + "-tutorial-login-sorted";
    const TString sortedIsRobotTable = "//tmp/" + GetUsername() + "-tutorial-is_robot-sorted";
    const TString humanTable = "//tmp/" + GetUsername() + "-tutorial-humans";
    const TString robotTable = "//tmp/" + GetUsername() + "-tutorial-robots";

    client->Sort(
        TSortOperationSpec()
            .AddInput("//home/tutorial/staff_unsorted")
            .Output(sortedLoginTable)
            .SortBy({"uid"}));

    client->Sort(
        TSortOperationSpec()
            .AddInput("//home/tutorial/is_robot_unsorted")
            .Output(sortedIsRobotTable)
            .SortBy({"uid"}));

    client->Reduce(
        TReduceOperationSpec()
            .ReduceBy({"uid"})
            .AddInput<TNode>(sortedLoginTable)
            .AddInput<TNode>(sortedIsRobotTable)
            .AddOutput<TNode>(robotTable)  // выходная таблица номер 0
            .AddOutput<TNode>(humanTable), // выходная таблица номер 1
        new TSplitHumanRobotsReduce);

    Cout << "Robot table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << robotTable << Endl;
    Cout << "Human table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << humanTable << Endl;

    return 0;
}

Reduce with multiple input and output tables (Protobuf version)

To create separate tables for human users and robots, we need to rewrite the same reducer to protobuf. The table with human users will contain the login, email, and name fields. The table with robots will contain the login and uid fields. We need to create separate protobuf message types for these tables.

The example is located at yt/cpp/mapreduce/examples/tutorial/multiple_input_multiple_output_reduce_protobuf.

import "yt/yt_proto/yt/formats/extension.proto";

message TUserRecord {
    optional int64 uid = 1    [(NYT.column_name) = "uid"];
    optional string name = 2  [(NYT.column_name) = "name"];
    optional string login = 3 [(NYT.column_name) = "login"];
}

message TIsRobotRecord {
    optional int64 uid = 1    [(NYT.column_name) = "uid"];
    optional bool is_robot = 2 [(NYT.column_name) = "is_robot"];
}

message THumanRecord {
    optional string name = 1  [(NYT.column_name) = "name"];
    optional string email = 2 [(NYT.column_name) = "email"];
    optional string login = 3 [(NYT.column_name) = "login"];
}

message TRobotRecord {
    optional int64 uid = 1    [(NYT.column_name) = "uid"];
    optional string login = 2 [(NYT.column_name) = "login"];
}

#include <yt/cpp/mapreduce/examples/tutorial/multiple_input_multiple_output_reduce_protobuf/data.pb.h>

#include <yt/cpp/mapreduce/interface/client.h>
#include <yt/cpp/mapreduce/interface/config.h>

#include <util/stream/output.h>
#include <util/system/user.h>

using namespace NYT;

class TSplitHumanRobotsReduce
    // Обратите внимание наш редьюс работает с несколькими типами записей
    // как на вход так и на выход, поэтому мы используем ::google::protobuf::Message
    : public IReducer<
        TTableReader<::google::protobuf::Message>,
        TTableWriter<::google::protobuf::Message>>
{
public:
    void Do(TReader* reader, TWriter* writer) override {
        TUserRecord userRecord;
        bool isRobot = false;
        for (auto& cursor : *reader) {
            auto tableIndex = cursor.GetTableIndex();
            // Мы знаем номер таблицы и поэтому мы можем запросить конкретный тип protobuf'а в этом месте.
            // Тип protobuf сообщения передаётся шаблонным аргументом к методу `GetRow()'.
            if (tableIndex == 0) {
                userRecord = cursor.GetRow<TUserRecord>();
            } else if (tableIndex == 1) {
                const auto& isRobotRecord = cursor.GetRow<TIsRobotRecord>();
                isRobot = isRobotRecord.is_robot();
            } else {
                Y_ABORT();
            }
        }

        // В AddRow мы можем передавать как TRobotRecord так и THumanRecord.
        if (isRobot) {
            TRobotRecord robotRecord;
            robotRecord.set_uid(userRecord.uid());
            robotRecord.set_login(userRecord.login());
            writer->AddRow(robotRecord, 0);
        } else {
            THumanRecord humanRecord;
            humanRecord.set_name(userRecord.name());
            humanRecord.set_login(userRecord.login());
            humanRecord.set_email(userRecord.login() + "@yandex-team.ru");
            writer->AddRow(humanRecord, 1);
        }
    }
};
REGISTER_REDUCER(TSplitHumanRobotsReduce)

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");

    const TString sortedUserTable = "//tmp/" + GetUsername() + "-tutorial-user-sorted";
    const TString sortedIsRobotTable = "//tmp/" + GetUsername() + "-tutorial-is_robot-sorted";
    const TString humanTable = "//tmp/" + GetUsername() + "-tutorial-humans";
    const TString robotTable = "//tmp/" + GetUsername() + "-tutorial-robots";

    client->Sort(
        TSortOperationSpec()
            .AddInput("//home/tutorial/staff_unsorted")
            .Output(sortedUserTable)
            .SortBy({"uid"}));

    client->Sort(
        TSortOperationSpec()
            .AddInput("//home/tutorial/is_robot_unsorted")
            .Output(sortedIsRobotTable)
            .SortBy({"uid"}));

    client->Reduce(
        TReduceOperationSpec()
            .ReduceBy({"uid"})
            .AddInput<TUserRecord>(sortedUserTable)
            .AddInput<TIsRobotRecord>(sortedIsRobotTable)
            .AddOutput<TRobotRecord>(robotTable)
            .AddOutput<THumanRecord>(humanTable),
        new TSplitHumanRobotsReduce);

    Cout << "Robot table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << robotTable << Endl;
    Cout << "Human table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << humanTable << Endl;

    return 0;
}

Reading and writing tables

When writing a table you can overwrite existing data or append your data to the table. When reading a table you can read the entire table or ranges by row number or key. Learn more about reading and writing data and tables.

The example is located at yt/cpp/mapreduce/examples/tutorial/table_read_write_tnode.

#include <yt/cpp/mapreduce/interface/client.h>

#include <util/system/user.h>

using namespace NYT;

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");

    const TString table = "//tmp/" + GetUsername() + "-read-write";

    {
        // Просто пишем данные в таблицу, если таблица существует, её перезапишут.
        auto writer = client->CreateTableWriter<TNode>(table);

        TNode row;
        row["english"] = "one";
        row["russian"] = "один";
        writer->AddRow(row);

        row["english"] = "two";
        row["russian"] = "два";
        writer->AddRow(row);

        // Лучше не забывать звать метод Finish явно.
        // Он конечно позовётся в деструкторе writer'а, но если случатся какие-нибудь ошибки,
        // то у вас не будет возможности их поймать, потому что деструктор в случае ошибки абортит программу.
        writer->Finish();
    }
    {
        // Дописываем данные в конец таблицы,
        // Для этого необходимо выставить опцию Append.
        auto writer = client->CreateTableWriter<TNode>(TRichYPath(table).Append(true));

        TNode row;
        row["english"] = "three";
        row["russian"] = "три";
        writer->AddRow(row);

        writer->Finish();
    }
    {
        // Читаем всю таблицу
        auto reader = client->CreateTableReader<TNode>(table);
        Cout << "*** ALL TABLE ***" << Endl;
        for (auto& cursor : *reader) { // reader имеет тот же самый интерфейс, что и reader в джобах
            auto& row = cursor.GetRow();
            Cout << "russian: " << row["russian"].AsString() << "; " << "english: " << row["english"].AsString() << Endl;
        }
        Cout << Endl;
    }
    {
        // Читаем первые 2 строки.
        auto reader = client->CreateTableReader<TNode>(TRichYPath(table).AddRange(
                TReadRange()
                    .LowerLimit(TReadLimit().RowIndex(0))
                    .UpperLimit(TReadLimit().RowIndex(2)))); // читаем с 0й по 2ю строки, 2я невключительно

        Cout << "*** FIRST TWO ROWS ***" << Endl;
        for (auto& cursor : *reader) {
            auto& row = cursor.GetRow();
            Cout << "russian: " << row["russian"].AsString() << "; " << "english: " << row["english"].AsString() << Endl;
        }
        Cout << Endl;
    }
    {
        // Мы можем отсортировать таблицу и читать записи по ключам.
        client->Sort(TSortOperationSpec()
            .SortBy({"english"})
            .AddInput(table)
            .Output(table));

        auto reader = client->CreateTableReader<TNode>(TRichYPath(table).AddRange(
                TReadRange()
                    .Exact(TReadLimit().Key({"three"})))); // если нужен один ключ а не диапазон, можно использовать
                                                           // Exact вместо LowerLimit / UpperLimit,
                                                           // Key указывает, что мы ищем запись по ключу и работает
                                                           // только для сортированных таблиц.

        Cout << "*** EXACT KEY ***" << Endl;
        for (auto& cursor : *reader) {
            auto& row = cursor.GetRow();
            Cout << "russian: " << row["russian"].AsString() << "; " << "english: " << row["english"].AsString() << Endl;
        }
        Cout << Endl;
    }

    return 0;
}

Transferring states to jobs

Sometimes you may need to write an operation that takes certain arguments. For example, write a program that filters the source table by user name.

The example is located at yt/cpp/mapreduce/examples/tutorial/stateful_map_tnode.

#include <yt/cpp/mapreduce/interface/client.h>

#include <library/cpp/string_utils/levenshtein_diff/levenshtein_diff.h>

#include <util/stream/output.h>
#include <util/system/user.h>

using namespace NYT;

class TFilterMapper
    : public IMapper<TTableReader<TNode>, TTableWriter<TNode>>
{
public:
    Y_SAVELOAD_JOB(Pattern_, MaxDistance_); // Заклинание, которое говорит, какие переменные нужно передавать на сервер.

    TFilterMapper() = default; // У джобы обязательно должен быть конструктор по умолчанию

    TFilterMapper(TString pattern, double maxDistance)
        : Pattern_(std::move(pattern))
        , MaxDistance_(maxDistance)
    { }


    void Do(TReader* reader, TWriter* writer) override {
        for (auto& cursor : *reader) {
            const auto& row = cursor.GetRow();
            const auto& name = row["name"].AsString();
            if (NLevenshtein::Distance(name, Pattern_) <= MaxDistance_) {
                writer->AddRow(row);
            }
        }
    }

private:
    TString Pattern_;
    size_t MaxDistance_ = 0;
};
REGISTER_MAPPER(TFilterMapper)

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");

    const TString outputTable = "//tmp/" + GetUsername() + "-tutorial-output-statefull-map";

    client->Map(
        TMapOperationSpec()
        .AddInput<TNode>("//home/tutorial/staff_unsorted")
        .AddOutput<TNode>(outputTable),
        new TFilterMapper("Arkady", 2)); // Мы создаём объект TFilterMapper, и конструктор заполняет поля Pattern_ и MaxDistance_.
                                         // Библиотека сериализует поля, указанные в Y_SAVELOAD_JOB, и загружает их на сервер.
                                         // На сервере вызывается конструктор по умолчанию и сериализованные поля загружаются.

    Cout << "Output table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << outputTable << Endl;

    return 0;
}

MapReduce operation (Protobuf version)

YTsaurus has the MapReduce merged operation, which works a bit faster than Map + Sort + Reduce. Let's return to one of our first examples and once again get the statistics of how many times a name occurs in our table with users. Before we do that, we first need to standardize the names by converting all of them to lowercase so that people with the names ARCHIE and Archie are counted as one.

The example is located at yt/cpp/mapreduce/examples/tutorial/mapreduce_protobuf.

#include <yt/cpp/mapreduce/interface/client.h>

#include <yt/cpp/mapreduce/examples/tutorial/mapreduce_protobuf/data.pb.h>

#include <util/stream/output.h>
#include <util/system/user.h>
#include <util/charset/utf8.h>

using namespace NYT;

//
// Для того чтобы запустить операцию mapreduce, нам нужны обычные классы Mapper'а и Reducer'а
// (эти классы даже можно использовать в других местах в отдельных операциях Map/Reduce).
//

class TNormalizeNameMapper
    : public IMapper<
        TTableReader<TLoginRecord>,
        TTableWriter<TLoginRecord>>
{
public:
    void Do(TReader* reader, TWriter* writer) override
    {
        for (auto& cursor : *reader) {
            auto row = cursor.GetRow();
            row.set_name(ToLowerUTF8(TString(row.name())));
            writer->AddRow(row);
        }
    }
};
REGISTER_MAPPER(TNormalizeNameMapper)

class TCountNameReducer
    : public IReducer<
        TTableReader<TLoginRecord>,
        TTableWriter<TNameStatistics>>
{
public:
    void Do(TReader* reader, TWriter* writer) override
    { TNameStatistics result;
        ui64 count = 0;
        for (auto& cursor : *reader) {
            const auto& row = cursor.GetRow();
            if (!result.has_name()) {
                result.set_name(row.name());
            }
            ++count;
        }
        result.set_count(count);
        writer->AddRow(result);
    }
};
REGISTER_REDUCER(TCountNameReducer)

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");

    const TString outputTable = "//tmp/" + GetUsername() + "-tutorial-output";

    // Запуск операции MapReduce несильно отличается от запуска других операций.
    // Нам надо указать список ключей, по которым мы будем редьюсить
    // и два класса -- один Mapper и один Reducer.
    client->MapReduce(
        TMapReduceOperationSpec()
            .ReduceBy({"name"})
            .AddInput<TLoginRecord>("//home/tutorial/staff_unsorted")
            .AddOutput<TNameStatistics>(outputTable),
        new TNormalizeNameMapper,
        new TCountNameReducer);

    Cout << "Output table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << outputTable << Endl;

    return 0;
}

MapReduce operation (version with lambda expressions)

The task is the same as in the previous example.

The example is located at yt/cpp/mapreduce/examples/tutorial/mapreduce_lambda.

#include <yt/cpp/mapreduce/examples/tutorial/mapreduce_lambda/data.pb.h>

#include <yt/cpp/mapreduce/library/lambda/yt_lambda.h>

#include <util/stream/output.h>
#include <util/system/user.h>
#include <util/charset/utf8.h>

using namespace NYT;

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");

    // Эта таблица будет с некоторой вероятностью не отсортирована.
    // Если вместо MapReduce() использовать MapReduceSorted(), тогда
    // она будет дополнительно досортирована по "name" и это будет отражено в схеме
    const TString outputTable = "//tmp/" + GetUsername() + "-tutorial-output";

    MapReduce<TLoginRecord, TLoginRecord, TNameStatistics>(
        client,
        "//home/tutorial/staff_unsorted",
        outputTable,
        {"name"}, // список ключей, по которым мы будем редьюсить
        [](auto& src, auto& dst) { // mapper
            dst.set_name(ToLowerUTF8(TString(src.name())));
            return true;
        },
        [](auto& /*src*/, auto& dst) { // reducer
            // dst.set_count() не вызываем, т.к. когда по
            // "name" редьюсим, это поле уже заполнено.

            // так можно делать, т.к. конструктор протобуфа dst
            // проинициализирует .Count в 0 (по умолчанию):
            dst.set_count(dst.count() + 1);
        });

    Cout << "Output table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << outputTable << Endl;

    return 0;
}

Preparing an operation in a job class

You can set up the operation parameters by overloading the IJob::PrepareOperation method. For more information, see Description.

import "yt/yt_proto/yt/formats/extension.proto";

message TGrepperRecord
{
    optional string key = 1;
    optional bytes other = 2 [(NYT.flags) = OTHER_COLUMNS];
};

#include <yt/cpp/mapreduce/examples/tutorial/prepare_operation/grepper.pb.h>

#include <yt/cpp/mapreduce/interface/client.h>
#include <yt/cpp/mapreduce/interface/config.h>

#include <util/stream/output.h>
#include <util/system/user.h>

using namespace NYT;

// Этот маппер отдаёт все строки, значение в колонке `column` которых равно `pattern`.
class TGrepper
    : public IMapper<TTableReader<TGrepperRecord>, TTableWriter<TGrepperRecord>>
{
public:
    TGrepper() = default;
    TGrepper(TString column, TString pattern)
        : Column_(column)
        , Pattern_(pattern)
    { }

    void Do(TReader* reader, TWriter* writer) override
    {
        for (const auto& cursor : *reader) {
            auto row = cursor.GetRow();
            if (row.key() == Pattern_) {
                writer->AddRow(row);
            }
        }
    }

    void PrepareOperation(const IOperationPreparationContext& context, TJobOperationPreparer& preparer) const override
    {
        auto schema = context.GetInputSchema(/* tableIndex */ 0);
        // Выходная схема совпадает со входной во всём, кроме названия колонки, по которой ищем.
        for (auto& column : schema.MutableColumns()) {
            if (column.Name() == Column_) {
                column.Name("Key");
            }
        }
        preparer
            .InputColumnRenaming(/* tableIndex */ 0, {{Column_, "Key"}})
            .InputDescription<TGrepperRecord>(/* tableIndex */ 0)
            // Выключаем автовывод схемы, т.к. он ничего не даст (у нас есть поле, помеченное OTHER_COLUMNS).
            .OutputDescription<TGrepperRecord>(/* tableIndex */ 0, /* inferSchema */ false)
            .OutputSchema(/* tableIndex */ 0, schema);
    }

    Y_SAVELOAD_JOB(Column_, Pattern_);

private:
    TString Column_;
    TString Pattern_;
};
REGISTER_MAPPER(TGrepper)

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");

    // Выходная табличка у нас будет лежать в tmp и содержать имя текущего пользователя.
    const TString outputTable = "//tmp/" + GetUsername() + "-tutorial-grepped-emails";

    client->Map(
        new TGrepper("login", "lev"),
        "//home/tutorial/staff_unsorted_schematized",
        outputTable);

    Cout << "Output table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << outputTable << Endl;

    return 0;
}

Advanced level

Running multiple operations in parallel

To run multiple operations in parallel, use the TOperationTracker class.

The example is located at yt/cpp/mapreduce/examples/tutorial/operation_tracker.

#include <yt/cpp/mapreduce/interface/client.h>
#include <yt/cpp/mapreduce/library/operation_tracker/operation_tracker.h> // OperationTracker живёт в этой библиотеке

#include <util/stream/output.h>
#include <util/system/user.h>

using namespace NYT;

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");

    const TString sortedLoginTable = "//tmp/" + GetUsername() + "-tutorial-login-sorted";
    const TString sortedIsRobotTable = "//tmp/" + GetUsername() + "-tutorial-is_robot-sorted";
    const TString humanTable = "//tmp/" + GetUsername() + "-tutorial-humans";
    const TString robotTable = "//tmp/" + GetUsername() + "-tutorial-robots";

    TOperationTracker tracker;
    tracker.AddOperation(
        client->Sort(
            TSortOperationSpec()
                .AddInput("//home/tutorial/staff_unsorted")
                .Output(sortedLoginTable)
                .SortBy({"uid"}),
            TOperationOptions().Wait(false))); // Wait(false) говорит клиенту не ждать завершения операции

    tracker.AddOperation(
        client->Sort(
            TSortOperationSpec()
                .AddInput("//home/tutorial/is_robot_unsorted")
                .Output(sortedIsRobotTable)
                .SortBy({"uid"}),
            TOperationOptions().Wait(false)));

    tracker.WaitAllCompleted(); // Ждёт когда обе операции завершатся, бросит исключение в случае ошибок.

    return 0;
}

Reduce with enable_key_guarantee=false

Suppose you want to filter a table with URLs by regular expressions that are located in a separate table with hosts (each host has its own regular expression). And while you could run Reduce on the host column, real data is often arranged in such a way that some hosts have a disproportionately large number of URLs. In our example, URLs from the host https://www.youtube.com take up more than half of the table. The job responsible for processing such a host will take a disproportionately large amount of time (keys like https://www.youtube.com here are also called monster keys).

However, you don't need all the records with the same host to go to the same machine to solve the task. You can distribute the table with URLs among several machines: note that the host must be received along with the URL.

host1, url1
host1, url1
host1, url3
host1, url3
host1, url2
host1, url2
host1, url4
host1, url4
host1, url5
host1, url5
host2, url1
host2, url1
host2, url2
host2, url2
host3, url1
host3, url1
host1, data1
host1, data1
host2, data1
host2, data1
host3, data1
host3, data1
host1, data1
host1, data1
host1, url1
host1, url1
host1, url3
host1, url3
host1, url2
host1, url2
host1, url4
host1, url4
host2, url1
host2, url1
host2, url2
host2, url2
host3, url1
host3, url1
host2, data1
host2, data1
host3, data1
host3, data1
Table 1
Table 1
Table 2
Table 2
Job 1
Job 1
Job 2
Job 2
Job 3
Job 3
Reduce
Reduce

Layer 1
host1, url1
host1, url1
host1, url3
host1, url3
host1, url2
host1, url2
host1, url4
host1, url4
host1, url5
host1, url5
host2, url1
host2, url1
host2, url2
host2, url2
host3, url1
host3, url1
host1, data1
host1, data1
host2, data1
host2, data1
host3, data1
host3, data1
host1, data1
host1, data1
host1, url1
host1, url1
host1, url2
host1, url2
host2, url1
host2, url1
host2, url2
host2, url2
host3, url1
host3, url1
host2, data1
host2, data1
host3, data1
host3, data1
Table 1
Table 1
Table 2
Table 2
Job 1
Job 1
Job 3
Job 3
Job 4
Job 4
host1, data1
host1, data1
host1, url3
host1, url3
host1, url4
host1, url4
Job 2
Job 2
Урлы с одного хоста теперь обрабатываются несколькими джобами
Урлы с одного хоста теперь обрабатываются несколькими джобами
Эту таблицу отмечаем как foreign
Эту таблицу отмечаем как foreign

Reduce

enable_key_guarantee=false

The example is located at yt/cpp/mapreduce/examples/tutorial/join_reduce_tnode.

#include <yt/cpp/mapreduce/interface/client.h>

#include <library/cpp/regex/pcre/regexp.h>

#include <util/system/user.h>

using namespace NYT;

class TFilterVideoRegexp
    : public IReducer<TTableReader<TNode>, TTableWriter<TNode>>
{
public:
    void Do(TReader* reader, TWriter* writer) override
    {
        // Так же как и с обычным Reducer'ом в каждый вызов метода Do
        // придут записи с общим JoinBy ключом.
        TMaybe<TRegExMatch> regex;
        for (auto& cursor : *reader) {
            auto row = cursor.GetRow();
            if (cursor.GetTableIndex() == 0) { // таблица с хостами
                const auto videoRegexp = row["video_regexp"].AsString();

                // Дебажная печать, stderr можно будет посмотреть в web интерфейсе
                Cerr << "Processing host: " << row["host"].AsString() << Endl;
                if (!videoRegexp.empty()) {
                    regex = TRegExMatch(videoRegexp);
                }
            } else { // таблица с урлами
                if (regex && regex->Match(row["path"].AsString().c_str())) {
                    writer->AddRow(row);
                }
            }
        }
    }
};
REGISTER_REDUCER(TFilterVideoRegexp)

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");

    const TString outputTable = "//tmp/" + GetUsername() + "-tutorial-join-reduce";
    const TString outputTmpVideoTable = "//tmp/" + GetUsername() + "-tutorial-join-reduce-video-sorted";
    const TString outputTmpDocTable = "//tmp/" + GetUsername() + "-tutorial-join-reduce-doc-sorted";

    client->Sort(TSortOperationSpec()
        .SortBy({"host"})
        .AddInput("//home/tutorial/host_video_regexp")
        .Output(outputTmpVideoTable));

    client->Sort(TSortOperationSpec()
        .SortBy({"host"})
        .AddInput("//home/tutorial/doc_title")
        .Output(outputTmpDocTable));

    client->Reduce(
        TReduceOperationSpec()
        .JoinBy({"host"})
        .AddInput<TNode>(
            TRichYPath(outputTmpVideoTable)
            .Foreign(true)) // важно отметить хостовую таблицу как foreign
        .AddInput<TNode>(outputTmpDocTable)
        .AddOutput<TNode>(outputTable)
        .EnableKeyGuarantee(false),
        new TFilterVideoRegexp);

    Cout << "Output table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << outputTable << Endl;

    return 0;
}

Batch requests

You can execute light requests (create/delete a table, check its existence, etc.) in batches. Use this method if you need to run a large amount of uniform operations. Batch requests can save a lot of time.

The example is located at yt/cpp/mapreduce/examples/tutorial/batch_request.

#include <yt/cpp/mapreduce/interface/client.h>

#include <util/system/user.h>

using namespace NYT;

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");

    // Создаём batch запрос (это можно делать и из транзакции).
    auto request = client->CreateBatchRequest();

    // Добавляем запросы в batch
    NThreading::TFuture<bool> docTitleExists = request->Exists("//home/tutorial/doc_title");
    NThreading::TFuture<bool> unexistingTableExists = request->Exists("//home/tutorial/unexisting_table");

    const TString outputTable = "//tmp/" + GetUsername() + "-tutorial-test-batch";
    NThreading::TFuture<TNodeId> createResult = request->Create(outputTable, NT_TABLE);

    // Выполняем batch запрос.
    request->ExecuteBatch();

    // Проверяем результаты.
    Cout << "Table //home/tutorial/doc_title exists: " << docTitleExists.GetValue() << Endl;
    Cout << "Table //home/tutorial/unexisting_table exists: " << unexistingTableExists.GetValue() << Endl;

    try {
        // Следует проверять все результаты с помощью GetValue(),
        // т.к. отдельные запросы могут пофейлиться и тогда соответствующая TFuture будет содержать ошибку.
        //
        // Если запускать эту программу второй раз то Create пофейлится, потому что таблица уже существует.
        createResult.GetValue();
    } catch (const std::exception& ex) {
        Cerr << "Create " << outputTable << " failed: " << ex.what() << Endl;
    }

    return 0;
}

Delivering a table dump to a job as a file

In YTsaurus, you can deliver files to jobs (for example, as was described earlier). A less popular method is delivering a table dump to a job. This can prove useful when a job requires a map stored in a small (a few gigabytes) table. However, don't use this approach to load large tables: if you have a table that is dozens of gigabytes in size, it won't fit on a single node of the cluster where the jobs are running.

The example is located at yt/cpp/mapreduce/examples/tutorial/pass_table_as_file.

#include <yt/cpp/mapreduce/interface/client.h>

#include <util/generic/hash_set.h>
#include <util/stream/output.h>
#include <util/system/user.h>

using namespace NYT;

class TFilterRobotsMap
    : public IMapper<TTableReader<TNode>, TTableWriter<TNode>>
{
public:
    // Мы читаем дамп таблицы в методе Start, перед тем как обработать первую запись в Map-таблице.
    // Настройка доставки таблицы в джобу происходит в функции main.
    void Start(TWriter*) override {
        // Нам в джобу доставили файлик "robot_table" с дампом таблицы.
        TFileInput tableDump("robot_table");

        // Есть функция CreateTableReader<>, которая умеет создавать читателя из любого IInputStream'а.
        // Созданный читатель имеет интерфейс аналогичный другим читателям.
        auto reader = CreateTableReader<TNode>(&tableDump);
        for (auto& cursor : *reader) {
            const auto& curRow = cursor.GetRow();
            if (curRow["is_robot"].AsBool()) {
                RobotUids.insert(curRow["uid"].AsInt64());
            }
        }
    }

    void Do(TReader* reader, TWriter* writer) override {
        for (auto& cursor : *reader) {
            const auto& curRow = cursor.GetRow();
            if (RobotUids.contains(curRow["uid"].AsInt64())) {
                writer->AddRow(curRow);
            }
        }
    }

private:
    THashSet<i64> RobotUids;
};
REGISTER_MAPPER(TFilterRobotsMap)

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");

    const TString outputTable = "//tmp/" + GetUsername() + "-tutorial-robots";

    client->Map(
        TMapOperationSpec()
            .MapperSpec(
                TUserJobSpec()
                // Самое интересное -- мы просим YT доставить нам табличку в виде файла.
                .AddFile(
                    TRichYPath("//home/tutorial/is_robot_unsorted") // Тут указываем таблицу, которую нам надо доставить.
                    .Format("yson") // Это формат, в котором таблица будет прочитана, нам нужен yson, чтобы TNode-читатель в джобе смог прочитать файл.
                    .FileName("robot_table") // Это имя файла, с дампом таблицы.
                    // Тут же можно было бы указать фильтрацию по колонкам или фильтрацию по номерам строк таблицы
                    // с помощью соответсвтующих методов TRichYPath, но нам ничего этого не надо.
                ))
            .AddInput<TNode>("//home/tutorial/staff_unsorted")
            .AddOutput<TNode>(outputTable),
        new TFilterRobotsMap);

    Cout << "Output table: https://yt.yandex-team.ru/freud/#page=navigation&offsetMode=row&path=" << outputTable << Endl;
    return 0;
}

Writing and retrieving job statistics

A running operation gathers a lot of statistics (see the documentation).

Example of a simple Map with statistics collection: yt/cpp/mapreduce/examples/tutorial/job_statistics:

#include <yt/cpp/mapreduce/interface/client.h>
#include <util/system/user.h>

using namespace NYT;

class TComputeEmailsMapper
    : public IMapper<TTableReader<TNode>, TTableWriter<TNode>>
{
    public:
        void Do(TReader* reader, TWriter* writer) override
        {
            // Соберём несколько интересных статистик по сотрудникам
            i64 fortyTwoCount = 0;
            i64 shortLoginCount = 0;
            for (const auto& cursor : *reader) {
                const auto& row = cursor.GetRow();
                auto login = row["login"].AsString();
                TNode outRow;
                outRow["name"] = row["name"];
                outRow["email"] = login + "@yandex-team.ru";
                writer->AddRow(outRow);

                if (login.find("42") != TString::npos) {
                    ++fortyTwoCount;
                }
                if (login.size() <= 4) {
                    ++shortLoginCount;
                }
            }

            // Пользовательскую статистику можно писать по пути (имени)
            WriteCustomStatistics("names/with_42", fortyTwoCount);
            WriteCustomStatistics(TNode() // А можно в виде мапы
                ("names", TNode()
                    ("short", shortLoginCount)));
        }
};
REGISTER_MAPPER(TComputeEmailsMapper)

int main() {
    NYT::Initialize();

    auto client = CreateClient("freud");
    const TString outputTable = "//tmp/" + GetUsername() + "-tutorial-emails";

    auto operation = client->Map(
        TMapOperationSpec()
            .AddInput<TNode>("//home/tutorial/staff_unsorted")
            .AddOutput<TNode>(outputTable),
        new TComputeEmailsMapper);

    auto jobStatistics = operation->GetJobStatistics();
    // Выведем системные статистики
    Cout << "time/total: " << jobStatistics.GetStatisticsAs<TDuration>("time/total").Sum() << Endl;
    Cout << "data/input/row_count: " << jobStatistics.GetStatistics("data/input/row_count").Sum() << Endl;

    // NB: для получения пользовательских статистик используется другой метод
    Cout << "names/with_42: " << jobStatistics.GetCustomStatistics("names/with_42").Sum() << Endl;
    Cout << "names/short: " << jobStatistics.GetCustomStatistics("names/short").Sum() << Endl;
    return 0;
}

Basic work with dynamic tables

YTsaurus supports dynamic tables. You can perform basic operations with such tables by using the C++ wrapper.

Attention

Unlike the other examples, you will have to take additional steps to run this one:

  • By default, users don't have the permissions to create dynamic tables, so you first need to get permissions to create and mount dynamic tables on some cluster.
  • When running the example, pass the cluster name and the path to the test table (it must not exist). For more information, see the comments in the program text.

The example is located at yt/cpp/mapreduce/examples/tutorial/dyntable_get_insert.

#include <yt/cpp/mapreduce/interface/client.h>

#include <yt/cpp/mapreduce/util/wait_for_tablets_state.h> // for WaitForTabletsState

#include <util/stream/output.h>

using namespace NYT;

int main(int argc, const char** argv) {
    NYT::Initialize();

    //
    // Программа принимает аргументами имя кластера и путь к таблице, с которой она будет работать (таблица не должна существовать).
    //
    // По умолчанию у пользователей нет прав монтировать динамические таблицы, и перед запуском программы необходимо получить
    // такие права (права на монтирование таблиц) на каком-нибудь кластере YT.
    if (argc != 3) {
        Cerr << "Usage:\n";
        Cerr << '\t' << argv[0] << " <server-name> <path-to-dynamic-table>\n";
        Cerr << '\n';
        Cerr << "For example:\n";
        Cerr << '\t' << argv[0] << " freud //home/ermolovd/test-dyntable" << Endl;
        return 1;
    }

    auto client = CreateClient(argv[1]);
    const TString dynamicTablePath = argv[2];

    auto schema = TTableSchema()
        .AddColumn("key", EValueType::VT_INT64, ESortOrder::SO_ASCENDING)
        .AddColumn("value", EValueType::VT_STRING);

    // ВАЖНО: при создании динамической таблицы нам нужно
    //  - указать атрибут dynamic: true
    //  - указать схему
    // Это нужно сделать обязательно в вызове Create. Т.е. не получится сначала создать таблицу,
    // а потом проставить эти атрибуты.
    client->Create(
        dynamicTablePath, NT_TABLE,
        TCreateOptions()
        .Force(true)
        .Attributes(
            TNode()
                ("dynamic", true)
                ("schema", schema.ToNode())));

    // Для того чтобы начать работу с динамической таблицей, её необходимо "подмонтировать".
    //
    // Часто создание / монтирование / размонтирование таблиц делается отдельными административными скриптами,
    // а приложение просто расчитывает, что таблицы существуют и уже подмонтированы.
    //
    // Мы для полноты примера подмонтируем таблицу, а в конце её размонтируем.
    client->MountTable(dynamicTablePath);
    Cout << "Waiting tablets are mounted..." << Endl;

    // Функция MountTable (и UnmountTable) запускает асинхронный процесс монтирования,
    // который может занять значительное время (десятки секунд) для больших таблиц.
    // Нам необходимо дождаться завершения этого процесса.
    WaitForTabletsState(client, dynamicTablePath, TS_MOUNTED);

    // Вставлять значения в динтаблицу можно с помощью InsertRows
    client->InsertRows(dynamicTablePath, {
        TNode()("key", 1)("value", "один"),
        TNode()("key", 42)("value", "сорок два"),
        TNode()("key", 100500)("value", "стопятьсот"),
    });

    // Получать значения из динтаблицы можно с помощью LookupRows,
    // возвращает список найденных строчек.
    auto result = client->LookupRows(dynamicTablePath, {
        TNode()("key", 100500),
        TNode()("key", 42),
    });

    Cout << "====== LOOKUP RESULT ======" << Endl;
    for (const auto& row : result) {
        Cout << "key: " << row["key"].AsInt64() << " value: " << row["value"].AsString() << Endl;
    }
    Cout << "====== END LOOKUP RESULT ======" << Endl;

    client->UnmountTable(dynamicTablePath);
    Cout << "Waiting tablets are unmounted..." << Endl;
    WaitForTabletsState(client, dynamicTablePath, TS_UNMOUNTED);

    return 0;
}
Previous
Lambda expressions
Next
Go API