Apache Parquet - Data Warehousing İçin Kullanılır

Giriş

Kısaca şöyle

- Designed for efficient storage and processing of large datasets (analytics, data warehousing) especially in big data frameworks like Apache Spark and Apache Hive.

- Uses advanced compression techniques : Uses dictionary encoding and run-length encoding, to compress data efficiently. This reduces storage space requirements and speeds up data retrieval.

- Optimized for read-heavy workloads

Tarihçesi

Açıklaması şöyle

Apache Parquet (jointly developed by Twitter and Cloudera), widely used in Hadoop ecosystems like Pig, Spark, and Hive, is a favored file format for column storage. The format, which employs a binary representation, is language-agnostic. With its .parquet extension, Parquet is designed for the efficient storage of substantial data sets.

Satır (Row) Formatı Neden Kötü

Her satır farklı bir disk sektöründe depolanırsa veriye erişim daha verimsiz hale gelmeye başlıyor. Ancak analytic yapılan işlemlerde genellikle tüm satıra değil sadece belli sütunlara ihtiyaç duyuluyor. Açıklaması şöyle

However, it can be inefficient when dealing with analytics, where you often only need specific columns from a large dataset.

For example, imagine a table with 50 columns and millions of rows. If you’re only interested in analyzing 3 of those columns, a row-wise format would still require you to read all 50 columns for each row.

Sütun (Column) Formatı

Tüm sütunlar aynı disk sektöründe depolanıyor. Böylece sadece gerekli veri okunuyor.  Açıklaması şöyle. Ancak problem verinin güncellenmesi aşamasında oluşuyor

However, simply storing data in a columnar format has some downsides. The record write or update operation requires touching multiple column segments, resulting in numerous I/O operations. This can significantly slow the write performance, especially when dealing with large datasets.

In addition, when queries involve multiple columns, the database system must reconstruct the records from separate columns. The cost of this reconstruction increases with the number of columns involved in the query.

Hybrid Formatı

Açıklaması şöyle. 

The hybrid format combines the best of both worlds.

The format groups data into “row groups,” each containing a subset of rows. (horizontal partition.) Within each row group, data for each column is called a “column chunk.” (vertical partition)

In the row group, these chunks are guaranteed to be stored contiguously on disk.

In the past, I thought Parquet was purely a columnar format, and I’m sure many of you might think the same. To describe it more precisely, Parquet organizes data in a hybrid format behind the scenes.

Avro vs Parquet

Açıklaması şöyle

Avro and Parquet are both compact binary storage formats that require a schema to structure the data that is being encoded. The difference is that Avro stores data in row format and Parquet stores data in a columnar format.

In my experience, these two formats are pretty much interchangeable. In fact, Parquet natively supports Avro schemas i.e., you could send Avro data to a Parquet reader and it would work just fine.

Örnek

Şöyle yaparız

employee_id (int32)
name (string)
salary (double)
hire_date (timestamp)

Schema Evolution

Açıklaması şöyle. Yani sadece field type değiştirilemez.

(1) Adding new fields: Suppose we have a Parquet file containing data with the following schema: If we want to add a new field called "gender" to the schema, we can do so without having to rewrite the entire file.

(2) Modifying field types: Suppose we have a Parquet file containing data with the following schema: If we want to change the data type of the "age" field from int to long, we cannot do so without breaking schema compatibility. Because the field type has been changed, Parquet cannot read and write data to the file using the new schema without rewriting the entire file.

(3) Deleting fields: Suppose we have a Parquet file containing data with the following schema: If we want to delete the "gender" field from the schema, we can do so without having to rewrite the entire file. Parquet can read and write data to the file using the new schema without having to rewrite the entire file.

Predicate Pushdown

Açıklaması şöyle

Predicate pushdown is a technique used in Parquet and other columnar storage formats to improve query performance by filtering data before it is read from disk. When a query is executed on a Parquet file, the query engine can push down filters to the storage layer, which allows for faster query performance by reducing the amount of data that needs to be read from the disk.

The basic idea behind predicate pushdown is to push the filtering operation as close to the data as possible. Instead of reading the entire dataset from disk and then filtering it in memory, the query engine pushes the filter operation down to the storage layer, which applies the filter during the data read operation. This can significantly reduce the amount of data that needs to be read from the disk, which in turn reduces query execution time.

HttpComponents HttpGet Sınıfı

Giriş

Şu satırı dahil ederiz.

import org.apache.http.client.methods.HttpGet;

constructor

Şöyle yaparız. Burada org.apache.http.util.EntityUtils kullanılıyor

void getFlight() throws Exception {
  HttpClient client = HttpClients.createDefault();
  HttpGet request = new HttpGet("https://www.cleartrip.com");
  HttpResponse response = client.execute(request);
  HttpEntity entity = response.getEntity();

  String responseBody = EntityUtils.toString(entity);
  int statusCode = response.getStatusLine().getStatusCode();
  System.out.println(statusCode);
  System.out.println(responseBody);
}

setHeader metodu

Örnek
Şöyle yaparız. Burada try block içinde kullanılıyor.

try(CloseableHttpClient httpClient = HttpClients.createDefault()) {
  HttpGet httpGet = new HttpGet( url.toString() );
  httpGet.setHeader( "Authorization", String.format( "token %s", "<token>" ));
  httpGet.setHeader( "Accept", "application/vnd.github.v3.raw" );

  try (CloseableHttpResponse httpResponse = httpClient.execute(httpGet)) {
     HttpEntity entity = httpResponse.getEntity();

    if (entity != null) {
       FileUtils.copyInputStreamToFile( entity.getContent(), file);
    }
  }
}

setHeaders metodu

Örnek

Şöyle yaparız

String sEndpoint = "https://mydomain.com:9090/sample";
Map<String, String> headers = new HashMap<>();
headers.put(HttpHeaders.AUTHORIZATION, BEARER_TOKEN);

String body = null;
HttpGet get = new HttpGet(sEndpoint);
get.setHeaders(headers.entrySet()alo
  .stream()
  .map(entry -> new BasicHeader(entry.getKey(), entry.getValue()))
  .toArray(Header[]::new));
try (CloseableHttpClient httpClient = ...;
     CloseableHttpResponse response = httpClient.execute(get)) {
  body = EntityUtils.toString(response.getEntity(), Charset.defaultCharset());
}

HttpComponents CloseableHttpAsyncClient Sınıfı

Giriş

Şu satırı dahil ederiz. Soyut bir sınıftır

import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;

execute metodu

Örnek

Şöyle yaparız

import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;

CloseableHttpAsyncClient httpClient = HttpAsyncClients.createDefault();
httpClient.start();

HttpGet request = new HttpGet("https://api.example.com/data");

httpClient.execute(request, new FutureCallback<HttpResponse>() {
  @Override
  public void completed(HttpResponse response) {
    System.out.println("Request completed with status: " + response.getStatusLine());
    // Process the response here
  }

  @Override
  public void failed(Exception ex) {
    System.out.println("Request failed: " + ex.getMessage());
  }

  @Override
  public void cancelled() {
    System.out.println("Request cancelled.");
  }
});

// Do other tasks here while the request is being executed asynchronously

// Wait for the response and clean up
httpClient.close();

HttpComponents HttpAsyncClients Sınıfı

Giriş

Şu satırı dahil ederiz.

import org.apache.http.impl.nio.client.HttpAsyncClients;

Bu sınıf sayesinde CloseableHttpAsyncClient  yaratılır. Bu sınıf ile gelen he şey senkron çalışır. Asenkron çalışmak için bir başka kütüphane olan HttpClients kullanılır

Maven
Şu satırı dahil ederiz

<dependency>
  <groupId>org.apache.httpcomponents</groupId>
  <artifactId>httpasyncclient</artifactId>
  <version>4.1.4</version>
</dependency>

createDefault metodu

Örnek

Şöyle yaparız

CloseableHttpAsyncClient httpClient = HttpAsyncClients.createDefault();

Apache Calcite SqlDialect Sınıfı

Giriş

Şu satırı dahil ederiz

import org.apache.calcite.sql.SqlDialect;

Bu sınıftan kalıtan şöyle sınıflar var

org.apache.calcite.sql.dialect.MysqlSqlDialect

org.apache.calcite.sql.dialect.PostgresqlSqlDialect

quoteIdentifier metodu

Belirtilen tablo ismi ve sütun ismini backtick veya çift tırnak (double quote) ile escape eder. Şeklen şöyle. Burada MySQL sunucusu ANSI SQL modda çalıştırılıyor

Apache Calcite

Giriş

Calcite sanırım javacc-maven-plugin kullanıyor. Bu bir parser yaratıyor Açıklaması şöyle. FMPP, FreeMarker söz dizimini kullanıyor

To use FMPP with Apache Calcite, you typically define the templates, metadata, and configuration files in a designated directory or package within the project. The FMPP tool is then invoked, specifying the input files, output directory, and any required configuration. FMPP processes the templates, substitutes the placeholders with the provided metadata, and generates the desired output files.

Optimizers

Açıklaması şöyle

There are two optimizers provided within Calcite:

HepPlanner - This heuristic planner is a rules based optimizer which attempts to match a number of rules to the query that can improve the performance through a number of different methods. This is similar in function to the RBO provided within Oracle.

VolcanoPlanner - This is a cost based optimizer which iterates through different rules and applies them to the query in different combinations until it can find a plan with the most efficient cost. Not all different plan permutations can be established, the optimizer will stop after a certain number of iterations or if the cost ceases to improve during the runs.

Adapter

Maven

Şu satırı dahil ederiz. 

<dependency>
    <groupId>org.apache.calcite</groupId>
    <artifactId>calcite-core</artifactId>
    <version>1.26.0</version>
</dependency>
<dependency>
    <groupId>org.apache.calcite.avatica</groupId>
    <artifactId>avatica-core</artifactId>
    <version>1.17.0</version>
</dependency>

Apache Avatica JDBC için kullanılır. Açıklaması şöyle

Avatica is a framework for building database drivers.

RelBuilder Sınıfı - Relational Algebra Builder

config metodu

Örnek

Şöyle yaparız

// Build our connection
Connection connection = DriverManager.getConnection("jdbc:calcite:");

// Unwrap our connection using the CalciteConnection
CalciteConnection calciteConnection = connection.unwrap(CalciteConnection.class);

// Get a pointer to our root schema for our Calcite Connection
SchemaPlus rootSchema = calciteConnection.getRootSchema();

// Attach our Postgres Jdbc Datasource to our Root Schema
rootSchema.add("exampleSchema", JdbcSchema.create(rootSchema, "exampleSchema", 
  dataSource, null, null));

FrameworkConfig config = Frameworks.newConfigBuilder()
                .defaultSchema(rootSchema)
                .build();

RelBuilder r = RelBuilder.create(config);

equals metodu

İki tane field alır

Örnek

Şöyle yaparız

//SELECT * FROM "cats" 
//    LEFT JOIN "dogs" ON "cats"."petID" = "dogs"."petID";

RelBuilder relBuilder = ...;
 
relBuilder.scan("petDB", "cats")
    .scan("petDB", "dogs").as("dogTableAlias")
    .join(JoinRelType.LEFT,
        relBuilder.equals(
            relBuilder.field(2, 0, "petID"),
            relBuilder.field(2, "dogTableAlias", "petID")
        )
);

Örnek

Şöyle yaparız. Burada sanırım iki farklı veri tabanından veri çekilip bunlar birleştiriliyor

import org.apache.calcite.adapter.jdbc.JdbcSchema;
import org.apache.calcite.interpreter.Bindables;
import org.apache.calcite.jdbc.CalciteConnection;
import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.plan.hep.HepPlanner;
import org.apache.calcite.plan.hep.HepProgram;
import org.apache.calcite.rel.RelHomogeneousShuttle;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.RelShuttle;
import org.apache.calcite.rel.core.TableScan;
import org.apache.calcite.rel.logical.LogicalTableScan;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.tools.FrameworkConfig;
import org.apache.calcite.tools.Frameworks;
import org.apache.calcite.tools.RelBuilder;
import org.apache.calcite.tools.RelRunner;
import org.verdictdb.commons.DBTablePrinter;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class CalciteJdbcExample {

    private static final String POSTGRESQL_SCHEMA = "PUBLIC";
    private static final String MYSQL_SCHEMA = "mysql";

    public static void main(String[] args) throws Exception {

        // Build our connection
        Connection connection = DriverManager.getConnection("jdbc:calcite:");

        // Unwrap our connection using the CalciteConnection
        CalciteConnection calciteConnection = connection.unwrap(CalciteConnection.class);

        // Get a pointer to our root schema for our Calcite Connection
        SchemaPlus rootSchema = calciteConnection.getRootSchema();

        // Instantiate a data source, this can be autowired in using Spring as well
        DataSource postgresDataSource = JdbcSchema.dataSource(
                "jdbc:postgresql://localhost/db",
                "org.postgresql.Driver", // Change this if you want to use something like MySQL, Oracle, etc.
                "postgres", // username
                "example"   // password
        );

        // Instantiate a data source, this can be autowired in using Spring as well
        DataSource mysqlDataSource = JdbcSchema.dataSource(
                "jdbc:mysql://localhost/db",
                "com.mysql.jdbc.Driver", // Change this if you want to use something like MySQL, Oracle, etc.
                "Username", // username
                "Password"   // password
        );

        // Attach our Postgres Jdbc Datasource to our Root Schema
        rootSchema.add(POSTGRESQL_SCHEMA, JdbcSchema.create(rootSchema, POSTGRESQL_SCHEMA, postgresDataSource, null, null));

        // Attach our MySQL Jdbc Datasource to our Root Schema
        rootSchema.add(MYSQL_SCHEMA, JdbcSchema.create(rootSchema, MYSQL_SCHEMA, mysqlDataSource, null, null));


        // Build a framework config to attach to our Calcite Planners and  Optimizers
        FrameworkConfig config = Frameworks.newConfigBuilder()
                .defaultSchema(rootSchema)
                .build();

        RelBuilder rb = RelBuilder.create(config);

        RelNode node = rb
                // First parameter is the Schema, the second is the table name
                .scan("PUBLIC", "TABLE_NAME_IN_POSTGRES")
                .scan("mysql", "TABLE_NAME_IN_MYSQL")
                // If you want to select from more than one table, you can do so by adding a second scan parameter
                .filter(
                        rb.equals(rb.field("fieldname"), rb.literal("literal"))
                )
                // These are the fields you want to return from your query
                .project(
                        rb.field("id"),
                        rb.field("col1"),
                        rb.field("colb")
                )
                .build();


        HepProgram program = HepProgram.builder().build();
        HepPlanner planner = new HepPlanner(program);

        planner.setRoot(node);

        RelNode optimizedNode = planner.findBestExp();

        final RelShuttle shuttle = new RelHomogeneousShuttle() {
            @Override public RelNode visit(TableScan scan) {
                final RelOptTable table = scan.getTable();
                if (scan instanceof LogicalTableScan && Bindables.BindableTableScan.canHandle(table)) {
                    return Bindables.BindableTableScan.create(scan.getCluster(), table);
                }
                return super.visit(scan);
            }
        };

        optimizedNode = optimizedNode.accept(shuttle);

        final RelRunner runner = connection.unwrap(RelRunner.class);
        PreparedStatement ps = runner.prepare(optimizedNode);

        ps.execute();

        ResultSet resultSet = ps.getResultSet();
        DBTablePrinter.printResultSet(resultSet);
    }
}