Тест

 
 CREATE TABLE if not exists test
 (
     id bigserial,
     amount int,
     summa  numeric(10,2),
     name varchar(1000),
     datetime timestamp default now()
 ) ;
 
 

 
 create table if not exists test(
     id integer,
     amount integer,
     summa Float64,
     name String ,
     datetime DateTime DEFAULT now())
 ENGINE = MergeTree()
 ORDER BY (id)
 SETTINGS index_granularity = 8192
 
 

 
 import psycopg2
 
 from datetime import datetime, timedelta
 import datetime as dt
 import os
 from io import StringIO
 from numpy.random import randint
 import string
 import random
 
 data = []
 
 def get_connect():
     conn = psycopg2.connect(dbname='postgres', user='postgres', host='localhost')
     return conn
 
 def create_table(connect, table_name):
     connect.set_isolation_level(0)
     cursor = connect.cursor()
     try:
         try:
             _str = '''drop TABLE %s''' % table_name
             cursor.execute(_str)
         except:
             pass    
 
         _str = '''
         CREATE TABLE if not exists test
     (
         id bigserial,
         amount int,
         summa  numeric(10,2),
         name varchar(1000),
         datetime timestamp default now()
     ) ;'''
         cursor.execute(_str)
 
         _str = 'ALTER TABLE test REPLICA IDENTITY FULL;' 
         cursor.execute(_str)
 
     except Exception as ex:
         print('... create_table error:', ex)
         pass    
 
     cursor.close()
 
 
 def select_sql(connect, _sql, table_name):
     connect.set_isolation_level(0)
     cursor = connect.cursor()
     cursor.execute(_sql)
     for f in cursor.fetchall():
         print(table_name,  f)
     cursor.close()
 
 def select_sql_test(connect, _sql, table_name):
     connect.set_isolation_level(0)
     cursor = connect.cursor()
     cursor.execute(_sql)
     # while(True):pass # 3.5 Гб
     for f in cursor.fetchall():
         _data = []
         _data.append(f[0])
         _data.append(f[1])
         _data.append(f[2])
         _data.append(f[3])
         _data.append(f[4])
         data.append(_data)
         #print(_data)
     #while(True):pass # 9.5 Гб
     cursor.close()
 
 def bulk_copy(f, cursor, table_name, _columns):
     try:
         columns = tuple(_columns)
         cursor.copy_from(f, table_name, columns = columns, sep=";", null='')
     except Exception as ex:
         print('... bulk_copy error:', str(ex))
 
 
 def insert(connect, table_name, columns, test_count):
     connect.set_isolation_level(0)
     cursor = connect.cursor()
 
 
     d_start = datetime.now()
 
     c = 0
     count = 1000
     _sql = ''
     _id = 0
     for i in range(0, count):
         _amount = randint(0, 1000000)
         _summa  = randint(0, 1000000)
         _name = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(100))
         _sql += str(_amount)
         _sql += ';'
         _sql += str(_summa)
         _sql += ';'
         _sql += str(_name)
         if table_name == 'test2':
             _sql += ';'
             _sql += str(_name)
         _sql += '''
 '''
 
     
     count2 = int(test_count / 1000)
     for i in range(0, count2):
         f = StringIO(_sql)
         bulk_copy(f, cursor, table_name, columns)
     cursor.close()
 
     select_sql(connect, 'select count(*) from %s' % table_name, table_name)
 
 
 connect = get_connect()
 print(connect)
 test_count = 10000000
 create_table(connect, 'test')
 datetime_start = datetime.now()
 insert(connect, 'test', ['amount', 'summa', 'name'], test_count)
 datetime_end = datetime.now()
 diff = (datetime_end - datetime_start).total_seconds()
 print('Время вставки %s записей: ' % test_count, diff)
 
 datetime_start = datetime.now()
 select_sql_test(connect, 'select * from test order by id', 'test')
 datetime_end = datetime.now()
 diff = (datetime_end - datetime_start).total_seconds()
 print('Время селекта %s записей: ' % test_count, diff)
 print('Размер прочитанного массива: ', len(data))    
 
 

 
 from clickhouse_driver import Client
 from datetime import datetime
 import random
 from numpy.random import randint
 import string
 
 
 def create_clickhouse(client):
     res = client.execute(' drop table if exists test')
     res = client.execute('''
     create table if not exists test(
         id integer,
         amount integer,
         summa Float64,
         name String ,
         datetime DateTime DEFAULT now()
     )
     ENGINE = MergeTree()
     ORDER BY (id)
     SETTINGS index_granularity = 8192
     ''')
     return res
 
 
 def insert_clickhouse(client, rows):
     try:
         _str =''' insert into test(
         amount,
         summa,
         name)
         values  '''
         res = client.execute(_str, rows)
     except Exception as ex:
         print('... insert_clickhouse error:', ex)
         for row in rows:
             _str = str(row)
 
 
 def select_clickhouse():
     client = Client(host='localhost', port='9000', user='default', database='default')
     data=[]
     date_start = datetime.now()
     res = client.execute('select * from test')
     # while True: pass # 5 Гб
     for f in res:
         _data = []
         _data.append(f[0])
         _data.append(f[1])
         _data.append(f[2])
         _data.append(f[3])
         _data.append(f[4])
         data.append(_data)
     date_end = datetime.now()
     _diff = (date_end - date_start).total_seconds()
     print("Время выборки 10 миллионов записей - %s секунд" %_diff)
     # while True: pass
 
 
 client = Client(host='localhost', port='9000', user='default', database='default')
 start = 1
 res  = create_clickhouse(client)
 count = 0
 rowcount = 0
 
 rows = []
 for i in range(0, 100000):
     _amount = randint(0, 1000000)
     _summa  = randint(0, 1000000)
     _name = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(100))
     _row = []
     _row.append(int(_amount))
     _row.append(float(_summa))
     _row.append(_name)
     rows.append(_row)
 
 date_start = datetime.now()
 count = 0 
 while True:
     res  = insert_clickhouse(client, rows)
     count += 1
     if count == 100:
         break
 
 date_end = datetime.now()
 _diff = (date_end - date_start).total_seconds()
 print("Время вставки 10 миллионов записей - %s секунд" %_diff)
 
 select_clickhouse()
 
 

 
 import java.sql.{Connection, DriverManager, ResultSet}
 import java.sql.PreparedStatement;
 import java.sql.*;
 import java.time.*;
 import java.io.InputStream
 import java.util.Properties;
 import scala.collection.mutable.ArrayBuffer
  
 class InsertToPostgres {
   classOf[org.postgresql.Driver]
   val con_str = "jdbc:postgresql://localhost:5432/postgres?user=postgres"
   val conn = DriverManager.getConnection(con_str)
 
   try {
       val stmt: Statement = conn.createStatement();
       val sql = "DROP TABLE if exists test;"
       stmt.executeUpdate(sql);
       val cstmt: Statement = conn.createStatement();
       val sql2 = """
             CREATE TABLE if not exists test
         (
             id bigserial,
             amount int,
             summa  numeric(10,2),
             name varchar(1000),
             datetime timestamp default now()
         ) ;"""
       cstmt.executeUpdate(sql2);
       val t1 = java.time.LocalDateTime.now;
 
       var count_batch = 0
       while(count_batch < 1000) {
           val ps: PreparedStatement = conn.prepareStatement("INSERT INTO test (amount, summa, name) VALUES (?,?,?)");
           var count = 0
           while(count < 10000) {
               ps.setInt(1, 1);
               ps.setFloat(2, 2);
               ps.setString(3,"12345лофврадлфоыраорkasjdgflsajdhflaksjdhflkajsdhflkajsdhljdfsdfsdfsdfsdfsdfsdfs");
               ps.addBatch();
               count += 1
           }
           ps.executeBatch();
           count_batch += 1
       }
 
       val t2 = java.time.LocalDateTime.now;
       var duration = Duration.between(t1, t2);
       System.out.printf("Время вставки 10 миллионов записей = %s секунд.%n", duration.getSeconds());    
 
    } finally {
      conn.close()
    }
 }
 
 
 class SelectFromPostgres {
   classOf[org.postgresql.Driver]
   val con_str = "jdbc:postgresql://localhost:5432/postgres?user=postgres"
   val conn = DriverManager.getConnection(con_str)
   try {
     val stm = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)
     val rs = stm.executeQuery("SELECT * from test")
     val t1 = java.time.LocalDateTime.now;
 
     var id     = ArrayBuffer[Int]()
     var amount = ArrayBuffer[Int]()
     var summa  = ArrayBuffer[Float]()
     var name   = ArrayBuffer[String]()
     var datetime = ArrayBuffer[java.sql.Timestamp]()
 
     while(rs.next) {
        id     += rs.getString("id").toInt
        amount += rs.getString("amount").toInt
        summa  += rs.getString("summa").toFloat
        name   += rs.getString("name")
        datetime += Timestamp.valueOf(rs.getString("datetime"))
     }
     val t2 = java.time.LocalDateTime.now;
     var duration = Duration.between(t1, t2);
     System.out.printf("Время выборки 10 миллионов записей = %s секунд.%n", duration.getSeconds());    
    } finally {
      conn.close()
    }
 }
 
 object pgconn extends App {
   //val Insert = new InsertToPostgres()
   val select = new SelectFromPostgres()
 
 }
 
 

 
 package com.clickhouse.examples.jdbc;
 
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.List;
 import java.util.ArrayList;
 import java.util.Random;
 import java.text.SimpleDateFormat;  
 
 public class Clickhouse {
     static void dropAndCreateTable(String url, String user, String password, String table) throws SQLException {
         try (Connection conn = DriverManager.getConnection(url, user, password);
                 Statement stmt = conn.createStatement()) {
             String stmt_execute = "drop table if exists test; " 
                                 + "create table if not exists test("
                                           +  "id integer,"
                                           +  "amount integer,"
                                           +  "summa Float64,"
                                           +  "name String ,"
                                           +  "datetime DateTime DEFAULT now())"
                                         +"ENGINE = MergeTree()"
                                         +"ORDER BY (id)"
                                         +"SETTINGS index_granularity = 8192";
 
             stmt.execute(stmt_execute);
         }catch (Exception e)
         {
             System.out.printf("... error: %s", e);
             //System.out.printf("... error: %s", e);
         }
 
     }
 
     static void batchInsert(String url, String user, String password, String table) throws SQLException {
         try (Connection conn = DriverManager.getConnection(url, user, password)) {
             // not that fast as it's based on string substitution and large sql statement
             String sql = String.format("insert into test (amount, summa, name) values(?, ?, ?)");
 
             java.time.LocalDateTime t1 = java.time.LocalDateTime.now();
 
             int count_batch = 0;
             while(count_batch < 1000) {
                 try (PreparedStatement ps = conn.prepareStatement(sql)) {
                       int count = 0;
                       while(count < 10000) {
                           ps.setInt(1, 1);
                           ps.setFloat(2, 2);
                           ps.setString(3,"12345лофврадлфоыраорkasjdgflsajdhflsdasdfasdfsdfsdfsdfsdfsdfsdfsdfsdfsdfs");
                           ps.addBatch();
                           count += 1;
                       }
                       ps.executeBatch();
                 }
                   count_batch += 1;
             }
             java.time.LocalDateTime t2 = java.time.LocalDateTime.now();
             java.time.Duration duration = java.time.Duration.between(t1, t2);
             System.out.printf("Время вставки 10 миллионов записей = %s секунд.%n", duration.getSeconds());    
 
         }
     }
 
     static int query(String url, String user, String password, String table) throws SQLException {
         try (Connection conn = DriverManager.getConnection(url, user, password);
                 Statement stmt = conn.createStatement();
                 ResultSet rs = stmt.executeQuery("select * from test")) {
             int count = 0;
 
             ArrayList id = new ArrayList();
             ArrayList amount = new ArrayList();
             ArrayList   summa = new ArrayList();
             ArrayList name = new ArrayList();
             //ArrayList datetime = new ArrayList();
             ArrayList datetime = new ArrayList();
 
             while (rs.next()) {
                 id.add(Integer.parseInt(rs.getString("id")));
                 amount.add(Integer.parseInt(rs.getString("amount")));
                 summa.add(Float.parseFloat(rs.getString("summa")));
                 name.add(rs.getString("name"));
                 datetime.add(rs.getString("datetime"));  
                 //datetime.add(java.time.LocalDateTime.parse(rs.getString("datetime")));  
                 count++;
             }
             return count;
         }
     }
 
     public static void main(String[] args) {
         String url = String.format("jdbc:ch://%s:%d/default", System.getProperty("chHost", "localhost"),
                 Integer.parseInt(System.getProperty("chPort", "8123")));
         String user = System.getProperty("chUser", "default");
         String password = System.getProperty("chPassword", "");
         String table = "test";
 
         try {
             dropAndCreateTable(url, user, password, table);
             batchInsert(url, user, password, table);
             java.time.LocalDateTime t1 = java.time.LocalDateTime.now();
             query(url, user, password, table);
             java.time.LocalDateTime t2 = java.time.LocalDateTime.now();
             java.time.Duration duration = java.time.Duration.between(t1, t2);
             System.out.printf("Время чтения 10 миллионов записей = %s секунд.%n", duration.getSeconds());    
         } catch (SQLException e) {
             e.printStackTrace();
         }
     }
 }
 
 

 
 package main
 
 import (
     "context"
     "time"
     "fmt"
     "os"
     "testing"
     "math/rand"
     "github.com/jackc/pgx/v4/pgxpool"
     "github.com/jackc/pgtype"
     "github.com/jackc/pgx/v4"
     "github.com/jackc/pgconn"
 )
 
 func mustConnectString(t testing.TB, connString string) *pgx.Conn {
     conn, err := pgx.Connect(context.Background(), connString)
     if err != nil {
         t.Fatalf("Unable to establish connection: %v", err)
     }
     return conn
 }
 
 func mustConnect(t testing.TB, config *pgx.ConnConfig) *pgx.Conn {
     conn, err := pgx.ConnectConfig(context.Background(), config)
     if err != nil {
         t.Fatalf("Unable to establish connection: %v", err)
     }
     return conn
 }
 
 func closeConn(t testing.TB, conn *pgx.Conn) {
     err := conn.Close(context.Background())
     if err != nil {
         t.Fatalf("conn.Close unexpectedly failed: %v", err)
     }
 }
 
 func mustExec(t testing.TB, conn *pgx.Conn, sql string, arguments ...interface{}) (commandTag pgconn.CommandTag) {
     var err error
     if commandTag, err = conn.Exec(context.Background(), sql, arguments...); err != nil {
         t.Fatalf("Exec unexpectedly failed with %v: %v", sql, err)
     }
     return
 }
 
 var t *testing.T
 
 func Insert() {
     databaseUrl := "postgres://postgres@localhost:5432/postgres"
     conn := mustConnectString(t, databaseUrl)
     defer closeConn(t, conn)
 
     sql := `DROP TABLE if exists test;;`
     mustExec(t, conn, sql)
     sql = `
         CREATE TABLE if not exists test
     (
         id bigserial,
         amount int,
         summa  numeric(10,2),
         name varchar(1000),
         datetime timestamp default now()
     ) ;`
     mustExec(t, conn, sql)
 
     batch := &pgx.Batch{}
 
     numInserts := 100000
 
     var letterRunes = []rune("bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUV")
 
     for i := 0; i < numInserts; i++ {
         _amount := rand.Int()/1000000000000
         _summa  := float32(rand.Int()/1000000000000)
         _name   := make([]rune, 100)
         for j := range _name {
             _name[j] = letterRunes[rand.Intn(len(letterRunes))]
         }
         batch.Queue("insert into test(amount, summa, name) values($1, $2, $3)", _amount, _summa , string(_name))
     }
 
     fmt.Printf("... start insert\n")
     t1 := time.Now()                                
 
     for i := 0; i < 100; i++ {
         res := conn.SendBatch(context.Background(), batch)
         res.Close()
     }
 
     t2 := time.Since(t1)                                
     fmt.Printf("... Время загрузки: =%v \n", t2.Seconds())
 
 }
 
 var pool *pgxpool.Pool
 
 func Select() {
 
         databaseUrl := "postgres://postgres@localhost:5432/postgres"
         dbPool, err := pgxpool.Connect(context.Background(), databaseUrl)
         if err != nil {
           fmt.Fprintf(os.Stderr, "Unable to connect to database: %v\n", err)
           os.Exit(1)
         }
         defer dbPool.Close()
 
         rows, err := dbPool.Query(context.Background(), "select * from test")
         if err != nil {
           fmt.Printf("error while executing query")
         }
 
         fmt.Printf("... start reading \n")
 
         _id := make([]int64, 0)
         _amount := make([]int32, 0)
         _summa := make([]pgtype.Numeric, 0)
         _name := make([]string, 0)
         _datetime := make([]time.Time, 0)
         
         t1 := time.Now()                                
         
         for rows.Next() {
           values, err := rows.Values()
           if err != nil {
             fmt.Printf("error while iterating dataset")
           }
           id := values[0].(int64)
           _id = append(_id, id)
           amount := values[1].(int32)
           _amount = append(_amount, amount)
           summa := values[2].(pgtype.Numeric)
           _summa = append(_summa, summa )
           name := values[3].(string)
           _name = append(_name, name )
           datetime := values[4].(time.Time)
           _datetime = append(_datetime, datetime )
         }
 
         t2 := time.Since(t1)                                
         fmt.Printf("... Время чтения: =%v число строк=%v\n", t2.Seconds(), len(_datetime))
 }
 
 func main(){
     Insert()
     Select() 
 }
 
 

 
 package main
 
 import (
     "context"
     "fmt"
     "log"
     "time"
     "math/rand"
     "github.com/ClickHouse/clickhouse-go/v2"
 )
 
 func Insert() error {
     var (
         ctx       = context.Background()
         conn, err = clickhouse.Open(&clickhouse.Options{
             Addr: []string{"127.0.0.1:9000"},
             Auth: clickhouse.Auth{
                 Database: "default",
                 Username: "default",
                 Password: "",
             },
             //Debug:           true,
             DialTimeout:     time.Second,
             MaxOpenConns:    10,
             MaxIdleConns:    5,
             ConnMaxLifetime: time.Hour,
         })
     )
     if err != nil {
         return err
     }
 
     err = conn.Exec(ctx, `drop table if  exists test`)
     if err != nil {
         return err
     }
 
     err = conn.Exec(ctx, `
         create table if not exists test(
             id integer,
             amount integer,
             summa Float64,
             name String ,
             datetime DateTime DEFAULT now()
         )
         ENGINE = MergeTree()
         ORDER BY (id)
         SETTINGS index_granularity = 8192
     `)
 
     if err != nil {
         return err
     }
 
     batch, err := conn.PrepareBatch(ctx, "INSERT INTO test(id, amount, summa, name) values")
     if err != nil {
         return err
     }
 
     var letterRunes = []rune("bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUV")
     count := 0
     _namej := ""
     for i := 0; i < 10000000; i++ {
         _amount := rand.Int()/1000000000000
         _summa  := float64(rand.Int()/1000000000000)
         _name   := make([]rune, 100)
         if count == 0 {
             for j := range _name {
                 _name[j] = letterRunes[rand.Intn(len(letterRunes))]
             }
             _namej = string(_name)
         }
         count += 1
         if count == 100 {count = 0}
         err := batch.Append(
             int32(i+1),
             int32(_amount),
             _summa,
             _namej,
         )
         if err != nil {
             return err
         }
     }
 
     t1 := time.Now()                                
     res := batch.Send()
     t2 := time.Since(t1)                                
     fmt.Printf("... Время вставки =%v число строк=10000000\n", t2.Seconds())
     return res
 }
 
 func Select() {
     conn, err := clickhouse.Open(&clickhouse.Options{
         Addr: []string{"127.0.0.1:9000"},
         Auth: clickhouse.Auth{
             Database: "default",
             Username: "default",
             Password: "",
         },
         Compression: &clickhouse.Compression{
             Method: clickhouse.CompressionLZ4,
         },
         Settings: clickhouse.Settings{
             "max_execution_time": 60,
         },
         //Debug: true,
     })
     if err := conn.Ping(context.Background()); err != nil {
         log.Fatal(err)
     }
 
     var settings []struct {
         Id     int32   `ch:"id"`
         Amount int32   `ch:"amount"`
         Summa  float64 `ch:"summa"`
         Name   string  `ch:"name"`
         Datetime *time.Time `ch:"datetime"`
     }
 
     t1 := time.Now()                                
 
     if err = conn.Select(context.Background(), &settings, "SELECT * FROM test "); err != nil {
         log.Fatal(err)
     }
 
     t2 := time.Since(t1)                                
     fmt.Printf("... Время выборки =%v число строк=10000000\n", t2.Seconds())
 
     // for{} 
 
 }
 
 func main() {
     Insert()
     Select()
 
 }
 
 
 

Оставьте свой комментарий !
Ваше имя: Комментарий: Оба поля являются обязательными  Автор   Комментарий к данной статье