public TpchDataSource(int sf, int part, int partCount, long minCustomerKey, long maxCustomerKey)
{
    super("tpch");
    Iterable<LineOrder> lineOrderIterable = TpchTable.LINEORDER.createGenerator(sf, part, partCount, minCustomerKey, maxCustomerKey);
    this.lineOrderIterator = lineOrderIterable.iterator();
    this.kryo = new Kryo();
    kryo.register(LineOrder.class, 10);
    kryo.register(byte[].class, 11);
    kryo.register(Object[].class, 12);
    this.output = new ByteBufferOutput(300, 2000);
}
 

/**
 * Serializes given object to byte buffer using Kryo instance in pool.
 *
 * @param obj    Object to serialize
 * @param buffer to write to
 */
public void serialize(final Object obj, final ByteBuffer buffer) {
  ByteBufferOutput out = new ByteBufferOutput(buffer);
  Kryo kryo = borrow();
  try {
    kryo.writeClassAndObject(out, obj);
    out.flush();
  } finally {
    release(kryo);
  }
}
 

/**
 * Serializes given object to OutputStream using Kryo instance in pool.
 *
 * @param obj        Object to serialize
 * @param stream     to write to
 * @param bufferSize size of the buffer in front of the stream
 */
public void serialize(final Object obj, final OutputStream stream, final int bufferSize) {
  ByteBufferOutput out = new ByteBufferOutput(stream, bufferSize);
  Kryo kryo = borrow();
  try {
    kryo.writeClassAndObject(out, obj);
    out.flush();
  } finally {
    release(kryo);
  }
}
 

/**
 * Serializes given object to byte buffer using Kryo instance in pool.
 *
 * @param obj Object to serialize
 * @param buffer to write to
 */
public void serialize(final Object obj, final ByteBuffer buffer) {
    ByteBufferOutput out = new ByteBufferOutput(buffer);
    Kryo kryo = borrow();
    try {
        kryo.writeClassAndObject(out, obj);
        out.flush();
    } finally {
        release(kryo);
    }
}
 

/**
 * Serializes given object to OutputStream using Kryo instance in pool.
 *
 * @param obj Object to serialize
 * @param stream to write to
 * @param bufferSize size of the buffer in front of the stream
 */
public void serialize(final Object obj, final OutputStream stream, final int bufferSize) {
    ByteBufferOutput out = new ByteBufferOutput(stream, bufferSize);
    Kryo kryo = borrow();
    try {
        kryo.writeClassAndObject(out, obj);
        out.flush();
    } finally {
        release(kryo);
    }
}
 

public KryoSerialization (Kryo kryo) {
	this.kryo = kryo;

	kryo.register(RegisterTCP.class);
	kryo.register(RegisterUDP.class);
	kryo.register(KeepAlive.class);
	kryo.register(DiscoverHost.class);
	kryo.register(Ping.class);

	input = new ByteBufferInput();
	output = new ByteBufferOutput();
}
 

@Test
public void testParquetCompression()
{
    final Kryo kryo = new Kryo();
    kryo.register(LineOrder.class);
    kryo.register(byte[].class);
    kryo.register(Object[].class);

    final LoaderConfig config = LoaderConfig.INSTANCE();
    try {
        config.init();
    }
    catch (ConfigFileNotFoundException e) {
        e.printStackTrace();
    }
    final MetaClient metaClient = new MetaClient("127.0.0.1", 10012);
    final int capacity = 8;
    Iterable<LineOrder> lineOrderIterable = TpchTable.LINEORDER.createGenerator(1000, 1, 1500, 0, 10000000);
    Iterator<LineOrder> lineOrderIterator = lineOrderIterable.iterator();
    TpchDataTransformer transformer = new TpchDataTransformer();
    ParaflowRecord[][] content = new ParaflowRecord[1][];
    ParaflowRecord[] records = new ParaflowRecord[capacity];
    int counter = 0;
    long textSize = 0;
    Output output = new ByteBufferOutput(300, 2000);
    while (lineOrderIterator.hasNext() && counter < capacity) {
        LineOrder lineOrder = lineOrderIterator.next();
        kryo.writeObject(output, lineOrder);
        textSize += output.position();
        ParaflowRecord record = transformer.transform(output.toBytes(), 80);
        records[counter++] = record;
        output.reset();
    }
    content[0] = records;
    output.close();
    ParaflowSegment segment = new ParaflowSegment(content, new long[0], new long[0], 0.0d);
    segment.setPath("file:///Users/Jelly/Desktop/1");
    MetaProto.StringListType columnNames = metaClient.listColumns("test", "debug01");
    MetaProto.StringListType columnTypes = metaClient.listColumnsDataType("test", "debug01");
    final ParquetSegmentWriter segmentWriter = new ParquetSegmentWriter(segment, metaClient, null);
    long start = System.currentTimeMillis();
    if (segmentWriter.write(segment, columnNames, columnTypes)) {
        System.out.println("Binary size: " + (1.0 * textSize / 1024.0 / 1024.0) + " MB.");
    }
    long end = System.currentTimeMillis();
    System.out.println("Time cost: " + (end - start));
}
 

@Override
public byte[] serialize(String s, SensorReading sensorReading) {
    ByteBufferOutput output = new ByteBufferOutput(100);
    kryos.get().writeObject(output, sensorReading);
    return output.toBytes();
}
 

@Override protected ByteBufferOutput initialValue() {
    return new ByteBufferOutput(10240);
}
 

@Override protected ByteBufferOutput initialValue() {
    return new ByteBufferOutput(10240);
}