clarksun

def chunks(l, n):
  for i in range(0, len(l), n):
    yield l[i:i+n]
    
d = dict()
for chunk in chunks(list(d.items()), 32768):
  # 因为batch用了transaction就不能用batch_size
  with table.batch(transaction=True) as b:
    for k, v in chunk:
      b.put(...)

clarksun

# https://pymotw.com/3/struct/
# struct_pack.py
import struct
import binascii

values = (1, 'ab'.encode('utf-8'), 2.7)
s = struct.Struct('I 2s f')
packed_data = s.pack(*values)

print('Original values:', values)

clarksun

# frontera hbase backend
from time import time
random_str = int(time() * 1E+6)  # 16位数字， 微秒， microsecond
random_str = int(time() * 1E+3)  # 13位数字， 毫秒,  millisecond

clarksun

# frontera HBaseQueue class

# tablename一定要to_bytes一下
from w3lib.util import to_bytes
table_name = to_bytes(table_name)
tables = set(connection.tables())
if drop and table_name in tables:
  connection.delete_table(table_name, disable=True)
  tables.remove(table_name)
  

clarksun

# frontera hbase backend _schedule()

table = self.connection.table(self.table_name)
with table.batch(transaction=True) as b:
  for rk, tuples in six.iteritems(data):
    obj = dict()
    for score, item in tuples:
      column = 'f:%0.3f_%0.3f' % get_interval(score, 0.001)
      obj.setdefault(column, []).append(item)
    

clarksun

# A simple log filter to turn debug logging on and off on a per-module
# basis, when using tornado-style logging.  Note that this works based
# on the module where the log statement occurs, not the name passed to
# logging.getLogger(), so it works even with libraries write directly
# to the root logger.
#
# One drawback to this approach (as opposed to using distinct
# per-module loggers and setting their levels appropriately) is that
# logger.isEnabledFor(DEBUG) will return true even when called from a
# module where debug output is not enabled, so modules that perform

clarksun

import gevent
import random
import time
from functools import wraps

# http://sdiehl.github.io/gevent-tutorial/

def task(pid):
  """
  Some non-deterministic task

clarksun

# kafka/consumer/group.py

class KafkaConsumer(six.Iterator):
  DEFAULT_CONFIG = {
    'bootstrap_servers': 'localhost',
    'xxx':'yyy',
  }
  
  def __init__(self, *topics, **configs):
    self.config = copy.copy(self.DEFAULT_CONFIG)

clarksun

class CaselessDict(dict):
  __slots__ = ()
  
  def __init__(self, seq=None):
    super(CaselessDict, self).__init__()
    if seq:
      self.update(seq)
      
  def __getitem__(self, key):
    return dict.__getitem__(self, self.normkey(key))

clarksun

kafka-run-class kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic xxx --time -1 --offsets 1 | awk -F ':' '{sum += $3} END {print sum}'