karmi · January 1, 2016 17:59
diff --git a/.gitignore b/.gitignore
 tmp/
diff --git a/01_prepare.rb b/01_prepare.rb
 require 'elasticsearch'

 INDEX  = 'random_data'
 CLIENT = Elasticsearch::Client.new trace: true

 CLIENT.indices.delete index: INDEX rescue nil

 CLIENT.indices.create index: INDEX,
                      body: {
                        settings: { index: {
                                      number_of_shards: 1,
                                      number_of_replicas: 0,
                                      refresh_interval: '-1',
                                      warmer: { enabled: false }
                                    }
                                  },
                        mappings: { d: {  dynamic: 'false',
                                          properties: {
                                            value:           { type: 'float' },
                                            created_on:      { type: 'date' },
                                            created_on_year: { type: 'short' },
                                            created_on_day:  { type: 'short' },
                                            created_on_hour: { type: 'short' }
                                          }
                                        }
                                  }
                      }

 CLIENT.cluster.health wait_for_status: 'green'

 CLIENT.indices.put_warmer index: INDEX,
                          name: 'aggregations',
                          body: {
                            query: {
                              filtered: {
                                filter: {
                                  range: {
                                    created_on: {
                                      gte: '2012-07-16T00:00:00Z',
                                      lte: '2012-07-16T01:00:00Z'
                                    }
                                  }
                                }
                              }
                            },
                            aggregations: {
                              years: {
                                terms: {
                                  field: "created_on_year",
                                  size: 1000,
                                  order: { _term: 'asc' }
                                },
                                aggregations: {
                                  stats: {
                                    stats: { field: 'value' }
                                  }
                                }
                              },

                              days: {
                                terms: {
                                  field: "created_on_day",
                                  size: 1000,
                                  order: { _term: 'asc' }
                                },
                                aggregations: {
                                  stats: {
                                    stats: { field: 'value' }
                                  }
                                }
                              },

                              hours: {
                                terms: {
                                  field: "created_on_hour",
                                  size: 1000,
                                  order: { _term: 'asc' }
                                },
                                aggregations: {
                                  stats: {
                                    stats: { field: 'value' }
                                  }
                                }
                              }
                            },
                            sort:  [ "created_on" ]
                          }
diff --git a/02_load.rb b/02_load.rb
 require 'logger'
 require 'curb'
 require 'csv'
 require 'oj'

 require 'elasticsearch'
 require 'elasticsearch/transport/transport/http/curb'

 if ARGV.empty?
  puts "[!] USAGE: ", "    #{__FILE__} [FILE]", "    cat <FILES> | #{__FILE__} -"
  exit(1)
 end

 at_exit do
  if $!.class == CSV::MalformedCSVError
    puts "", "[!] Seems like you have hit the #{$!.class} error, you have to fix the CSV first with eg.:",
             '    > cat randomData10.csv | tr -d \'\r\' > randomData10_fixed.csv', ""
  end
 end

 =begin
 NOTE: It might make sense to parallelize the loading of huge CSV file by splitting it
      into smaller files and loading them separately, eg.:

      > export FILENAME=file.txt
      > tail -n +2 $FILENAME | split -l 100000 -a 4 - split_
      > for file in split_*
        do
          head -n 1 $FILENAME > tmp_file
          cat $file >> tmp_file
          mv -f tmp_file $file
        done

      > for file in split_*; do (time ruby 02_load.rb $file)& done
 =end

 INDEX  = ENV['INDEX'] || 'random_data'
 BATCH  = 1_000

 csv    = CSV.new(ARGF, headers: true, skip_blanks: true)
 buffer = []
 count  = 0

 CLIENT = Elasticsearch::Client.new transport_class: Elasticsearch::Transport::Transport::HTTP::Curb,
                                   logger: ( l = Logger.new(STDERR); l.level = Logger::INFO; l )

 def send_batch(buffer)
  CLIENT.bulk index: INDEX,
              type: 'd',
              body: buffer.map { |item| { index: { data: item } } }
 end

 csv.each do |row|
  if count > 0 && (count % BATCH) == 0
    send_batch buffer
    buffer = []
  end

  hash = row.to_hash

  document = {
    value:      hash['value'],
    created_on: hash['created_on'],
    created_on_year: Time.parse(hash['created_on']).year,
    created_on_day:  Time.parse(hash['created_on']).yday,
    created_on_hour: Time.parse(hash['created_on']).hour
  }

  buffer << document
  count += 1
 end

 send_batch buffer unless buffer.empty?

 CLIENT.indices.put_settings index: INDEX, body: { refresh_interval: '1s', warmer: { enabled: true } }
 CLIENT.indices.refresh index: INDEX
diff --git a/03_aggregate.rb b/03_aggregate.rb
 require 'logger'
 require 'curb'
 require 'oj'
 require 'ansi/table'

 require 'elasticsearch'
 require 'elasticsearch/transport/transport/http/curb'

 INDEX  = 'random_data'

 CLIENT = Elasticsearch::Client.new transport_class: Elasticsearch::Transport::Transport::HTTP::Curb,
                                   trace: (ENV['VERBOSE'] =~ /y*|t*/)

 json = {
  query: {
    match_all: {}
  },
  aggregations: {
    years: {
      terms: {
        field: "created_on_year",
        size: 1000,
        order: { _term: 'asc' }
      },
      aggregations: {
        stats: {
          stats: { field: 'value' }
        }
      }
    },

    days: {
      terms: {
        field: "created_on_day",
        size: 1000,
        order: { _term: 'asc' }
      },
      aggregations: {
        stats: {
          stats: { field: 'value' }
        }
      }
    },

    hours: {
      terms: {
        field: "created_on_hour",
        size: 1000,
        order: { _term: 'asc' }
      },
      aggregations: {
        stats: {
          stats: { field: 'value' }
        }
      }
    }
  }
 }

 response = CLIENT.search index: INDEX,
                         search_type: 'count',
                         body:  json;


 puts

 puts "TOOK: #{response['took']}ms", '='*ANSI::Terminal.terminal_width

 response['aggregations'].each do |key, value|
  puts "#{key.upcase.center(10)}", "-"*ANSI::Terminal.terminal_width
  value['buckets'].each do |bucket|
    puts bucket['key'].to_s.center(10) + " " + bucket['stats'].map { |k, v|
           k =~ /count/ ? "#{k}: #{v.to_s.ljust(5)}" : "#{k}: #{v.to_f.to_s.ljust(25)}"
         }.join(" ")
  end
  puts
 end;
diff --git a/04_aggregate_and_filter.rb b/04_aggregate_and_filter.rb
 require 'logger'
 require 'curb'
 require 'oj'
 require 'ansi/table'

 require 'elasticsearch'
 require 'elasticsearch/transport/transport/http/curb'

 INDEX  = 'random_data'

 CLIENT = Elasticsearch::Client.new transport_class: Elasticsearch::Transport::Transport::HTTP::Curb,
                                   trace: (ENV['VERBOSE'] =~ /y*|t*/)

 json = {
  query: {
    filtered: {
      filter: {
        range: {
          created_on: {
            gte: '2012-07-16T00:00:00Z',
            lte: '2012-07-16T01:00:00Z'
          }
        }
      }
    }
  },
  aggregations: {
    years: {
      terms: {
        field: "created_on_year",
        size: 1000,
        order: { _term: 'asc' }
      },
      aggregations: {
        stats: {
          stats: { field: 'value' }
        }
      }
    },

    days: {
      terms: {
        field: "created_on_day",
        size: 1000,
        order: { _term: 'asc' }
      },
      aggregations: {
        stats: {
          stats: { field: 'value' }
        }
      }
    },

    hours: {
      terms: {
        field: "created_on_hour",
        size: 1000,
        order: { _term: 'asc' }
      },
      aggregations: {
        stats: {
          stats: { field: 'value' }
        }
      }
    }
  }
 }

 response = CLIENT.search index: INDEX,
                         search_type: 'count',
                         body:  json;


 puts

 puts "TOOK: #{response['took']}ms", '='*ANSI::Terminal.terminal_width

 response['aggregations'].each do |key, value|
  puts "#{key.upcase.center(10)}", "-"*ANSI::Terminal.terminal_width
  value['buckets'].each do |bucket|
    puts bucket['key'].to_s.center(10) + " " + bucket['stats'].map { |k, v|
           k =~ /count/ ? "#{k}: #{v.to_s.ljust(5)}" : "#{k}: #{v.to_f.to_s.ljust(25)}"
         }.join(" ")
  end
  puts
 end;
diff --git a/__readme.markdown b/__readme.markdown
diff --git a/data_10K.csv b/data_10K.csv
diff --git a/output.txt b/output.txt
 [1] First version

 Loading 50M in 50 processes:
 real  161m27.897s

 ruby scripts/04_aggregate_and_filter.rb
 TOOK: 943ms

 -------------------------------------------------------------------------------

 [2] Second version

 Loading + time parsing 50M in 4 processes:
 real  234m38.457s

 time ruby 03_aggregate.rb

 TOOK: 16458ms
 real  0m16.865s

 First run:

 ruby scripts/04_aggregate_and_filter.rb
 TOOK: 54082ms

 Second run:

 ruby scripts/04_aggregate_and_filter.rb
 TOOK: 26ms

 Subsequent runs:

 ruby scripts/04_aggregate_and_filter.rb
 TOOK: 10ms

 -------------------------------------------------------------------------------

 [3] Third version (warmers)

 time ruby 03_aggregate.rb

 TOOK: 16458ms
 real  0m16.865s

 ruby scripts/04_aggregate_and_filter.rb
 TOOK: 26ms
	require 'elasticsearch'

	INDEX = 'random_data'
	CLIENT = Elasticsearch::Client.new trace: true

	CLIENT.indices.delete index: INDEX rescue nil

	CLIENT.indices.create index: INDEX,
	body: {
	settings: { index: {
	number_of_shards: 1,
	number_of_replicas: 0,
	refresh_interval: '-1',
	warmer: { enabled: false }
	}
	},
	mappings: { d: { dynamic: 'false',
	properties: {
	value: { type: 'float' },
	created_on: { type: 'date' },
	created_on_year: { type: 'short' },
	created_on_day: { type: 'short' },
	created_on_hour: { type: 'short' }
	}
	}
	}
	}

	CLIENT.cluster.health wait_for_status: 'green'

	CLIENT.indices.put_warmer index: INDEX,
	name: 'aggregations',
	body: {
	query: {
	filtered: {
	filter: {
	range: {
	created_on: {
	gte: '2012-07-16T00:00:00Z',
	lte: '2012-07-16T01:00:00Z'
	}
	}
	}
	}
	},
	aggregations: {
	years: {
	terms: {
	field: "created_on_year",
	size: 1000,
	order: { _term: 'asc' }
	},
	aggregations: {
	stats: {
	stats: { field: 'value' }
	}
	}
	},

	days: {
	terms: {
	field: "created_on_day",
	size: 1000,
	order: { _term: 'asc' }
	},
	aggregations: {
	stats: {
	stats: { field: 'value' }
	}
	}
	},

	hours: {
	terms: {
	field: "created_on_hour",
	size: 1000,
	order: { _term: 'asc' }
	},
	aggregations: {
	stats: {
	stats: { field: 'value' }
	}
	}
	}
	},
	sort: [ "created_on" ]
	}
	require 'logger'
	require 'curb'
	require 'csv'
	require 'oj'

	require 'elasticsearch'
	require 'elasticsearch/transport/transport/http/curb'

	if ARGV.empty?
	puts "[!] USAGE: ", " #{__FILE__} [FILE]", " cat <FILES> \| #{__FILE__} -"
	exit(1)
	end

	at_exit do
	if $!.class == CSV::MalformedCSVError
	puts "", "[!] Seems like you have hit the #{$!.class} error, you have to fix the CSV first with eg.:",
	' > cat randomData10.csv \| tr -d \'\r\' > randomData10_fixed.csv', ""
	end
	end

	=begin
	NOTE: It might make sense to parallelize the loading of huge CSV file by splitting it
	into smaller files and loading them separately, eg.:

	> export FILENAME=file.txt
	> tail -n +2 $FILENAME \| split -l 100000 -a 4 - split_
	> for file in split_*
	do
	head -n 1 $FILENAME > tmp_file
	cat $file >> tmp_file
	mv -f tmp_file $file
	done

	> for file in split_*; do (time ruby 02_load.rb $file)& done
	=end

	INDEX = ENV['INDEX'] \|\| 'random_data'
	BATCH = 1_000

	csv = CSV.new(ARGF, headers: true, skip_blanks: true)
	buffer = []
	count = 0

	CLIENT = Elasticsearch::Client.new transport_class: Elasticsearch::Transport::Transport::HTTP::Curb,
	logger: ( l = Logger.new(STDERR); l.level = Logger::INFO; l )

	def send_batch(buffer)
	CLIENT.bulk index: INDEX,
	type: 'd',
	body: buffer.map { \|item\| { index: { data: item } } }
	end

	csv.each do \|row\|
	if count > 0 && (count % BATCH) == 0
	send_batch buffer
	buffer = []
	end

	hash = row.to_hash

	document = {
	value: hash['value'],
	created_on: hash['created_on'],
	created_on_year: Time.parse(hash['created_on']).year,
	created_on_day: Time.parse(hash['created_on']).yday,
	created_on_hour: Time.parse(hash['created_on']).hour
	}

	buffer << document
	count += 1
	end

	send_batch buffer unless buffer.empty?

	CLIENT.indices.put_settings index: INDEX, body: { refresh_interval: '1s', warmer: { enabled: true } }
	CLIENT.indices.refresh index: INDEX
	[1] First version

	Loading 50M in 50 processes:
	real 161m27.897s

	ruby scripts/04_aggregate_and_filter.rb
	TOOK: 943ms

	-------------------------------------------------------------------------------

	[2] Second version

	Loading + time parsing 50M in 4 processes:
	real 234m38.457s

	time ruby 03_aggregate.rb

	TOOK: 16458ms
	real 0m16.865s

	First run:

	ruby scripts/04_aggregate_and_filter.rb
	TOOK: 54082ms

	Second run:

	ruby scripts/04_aggregate_and_filter.rb
	TOOK: 26ms

	Subsequent runs:

	ruby scripts/04_aggregate_and_filter.rb
	TOOK: 10ms

	-------------------------------------------------------------------------------

	[3] Third version (warmers)

	time ruby 03_aggregate.rb

	TOOK: 16458ms
	real 0m16.865s

	ruby scripts/04_aggregate_and_filter.rb
	TOOK: 26ms