Java 类org.apache.avro.io.JsonDecoder 实例源码

static void convertJsonToAvro(InputStream inputStream, OutputStream outputStream, Schema schema)
        throws IOException {
    DatumReader<Object> reader = new GenericDatumReader<>(schema);
    DatumWriter<Object> writer = new GenericDatumWriter<>(schema);

    Encoder binaryEncoder = EncoderFactory.get().binaryEncoder(outputStream, null);

    JsonDecoder jsonDecoder = DecoderFactory.get().jsonDecoder(schema, inputStream);
    Object datum = null;
    while (true) {
        try {
            datum = reader.read(datum, jsonDecoder);
        } catch (EOFException eofException) {
            break;
        }
        writer.write(datum, binaryEncoder);
        binaryEncoder.flush();
    }
    outputStream.flush();
}

static YarnClusterSubmissionFromCS readYarnClusterSubmissionFromCSFromInputStream(
    final InputStream appInputStream, final InputStream jobInputStream) throws IOException {
  final JsonDecoder appDecoder = DecoderFactory.get().jsonDecoder(
      AvroYarnAppSubmissionParameters.getClassSchema(), appInputStream);
  final SpecificDatumReader<AvroYarnAppSubmissionParameters> appReader = new SpecificDatumReader<>(
      AvroYarnAppSubmissionParameters.class);
  final AvroYarnAppSubmissionParameters yarnClusterAppSubmissionParameters = appReader.read(null, appDecoder);

  final JsonDecoder jobDecoder = DecoderFactory.get().jsonDecoder(
      AvroYarnClusterJobSubmissionParameters.getClassSchema(), jobInputStream);
  final SpecificDatumReader<AvroYarnClusterJobSubmissionParameters> jobReader = new SpecificDatumReader<>(
      AvroYarnClusterJobSubmissionParameters.class);
  final AvroYarnClusterJobSubmissionParameters yarnClusterJobSubmissionParameters = jobReader.read(null, jobDecoder);

  return new YarnClusterSubmissionFromCS(yarnClusterAppSubmissionParameters, yarnClusterJobSubmissionParameters);
}

@Override
public RecordField generateFormDataFromJson(String avroSchema, String json)
        throws AvroUiSandboxServiceException {
    try {
        Schema schema = new Schema.Parser().parse(avroSchema);
        JsonDecoder jsonDecoder = DecoderFactory.get().jsonDecoder(schema, json);
        DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>(schema);
        GenericRecord genericRecord = datumReader.read(null, jsonDecoder);
        return FormAvroConverter.createRecordFieldFromGenericRecord(genericRecord);
    } catch (Exception e) {
        throw Utils.handleException(e);
    }
}

/**
 * Reads avro object from input stream.
 *
 * @param inputStream The input stream to read from
 * @return Avro object
 * @throws IOException
 */
AvroYarnJobSubmissionParameters fromInputStream(final InputStream inputStream) throws IOException {
  final JsonDecoder decoder = DecoderFactory.get().jsonDecoder(
          AvroYarnJobSubmissionParameters.getClassSchema(), inputStream);
  final SpecificDatumReader<AvroYarnJobSubmissionParameters> reader = new SpecificDatumReader<>(
          AvroYarnJobSubmissionParameters.class);
  return reader.read(null, decoder);
}

static AvroYarnAppSubmissionParameters readYarnAppSubmissionParametersFromInputStream(
    final InputStream inputStream) throws IOException {
  final JsonDecoder decoder = DecoderFactory.get().jsonDecoder(
      AvroYarnAppSubmissionParameters.getClassSchema(), inputStream);
  final SpecificDatumReader<AvroYarnAppSubmissionParameters> reader = new SpecificDatumReader<>(
      AvroYarnAppSubmissionParameters.class);
  return reader.read(null, decoder);
}

static AvroYarnJobSubmissionParameters readYarnJobSubmissionParametersFromInputStream(
    final InputStream inputStream) throws IOException {
  final JsonDecoder decoder = DecoderFactory.get().jsonDecoder(
      AvroYarnJobSubmissionParameters.getClassSchema(), inputStream);
  final SpecificDatumReader<AvroYarnJobSubmissionParameters> reader = new SpecificDatumReader<>(
      AvroYarnJobSubmissionParameters.class);
  return reader.read(null, decoder);
}

/**
 * Reads avro object from input stream.
 *
 * @param inputStream The input stream to read from
 * @return Avro object
 * @throws IOException
 */
AvroMultiRuntimeAppSubmissionParameters fromInputStream(final InputStream inputStream) throws IOException {
  final JsonDecoder decoder = DecoderFactory.get().jsonDecoder(
          AvroMultiRuntimeAppSubmissionParameters.getClassSchema(), inputStream);
  final SpecificDatumReader<AvroMultiRuntimeAppSubmissionParameters> reader = new SpecificDatumReader<>(
          AvroMultiRuntimeAppSubmissionParameters.class);
  return reader.read(null, decoder);
}

@SuppressWarnings("deprecation")
protected Decoder createDecoder() throws IOException {
  switch(codecType) {
    case BINARY:
      return new BinaryDecoder(getOrCreateInputStream());
    case JSON:
      return new JsonDecoder(schema, getOrCreateInputStream());
  }
  return null;
}

public <T extends SpecificRecordBase> List<T> getResultsSince(Endpoint ep, Class<T> cls,
        Schema schema, long since) {

    ArrayList<T> reslist = new ArrayList<T>();  

    DBCollection c  = db.getCollection(COLL_AVAIL);
    DBCursor curs = null;
    try{
    if(ep==null){
        curs = c.find();
    }else{

        DBObject q =
                QueryBuilder.start().and(
                        QueryBuilder.start(RESULT_KEY).is(ep.getUri().toString()).get(),
                        QueryBuilder.start("endpointResult.start").greaterThan(since).get()).get();
        log.info("[EXEC] {}",q);
        curs = c.find(q);
    }

    while(curs.hasNext()){
        DBObject o = curs.next();
        SpecificDatumReader r = new SpecificDatumReader<T>(cls);
        JsonDecoder d;
        try {
            d = DecoderFactory.get().jsonDecoder(schema, o.toString());
            T t =(T) r.read(null, d);
            reslist.add(t);
        } catch (IOException e) {
            log.error("GET RESULT Exception: {} {}", ep.getUri(), ExceptionHandler.logAndtoString(e,true));
        }
    }
    }finally{
        if(curs!=null)
            curs.close();
    }
    return reslist;

}

public <T extends SpecificRecordBase> List<T> getResultsSince(Endpoint ep, Class<T> cls,
        Schema schema, long from, long to) {
    ArrayList<T> reslist = new ArrayList<T>();  

    DBCollection c  = db.getCollection(COLL_AVAIL);
    DBCursor curs = null;
    try{
    if(ep==null){
        curs = c.find();
    }else{

        DBObject q =
                QueryBuilder.start().and(
                        QueryBuilder.start(RESULT_KEY).is(ep.getUri().toString()).get(),
                        QueryBuilder.start("endpointResult.start").greaterThan(from).get(),
                        QueryBuilder.start("endpointResult.start").lessThanEquals(to).get()).get();
        log.info("[EXEC] {}",q);
        curs = c.find(q);
    }

    while(curs.hasNext()){
        DBObject o = curs.next();
        SpecificDatumReader r = new SpecificDatumReader<T>(cls);
        JsonDecoder d;
        try {
            d = DecoderFactory.get().jsonDecoder(schema, o.toString());
            T t =(T) r.read(null, d);
            reslist.add(t);
        } catch (IOException e) {
            log.error("GET RESULT SINCE Exception: {} {}", ep.getUri(), ExceptionHandler.logAndtoString(e,true));
        }
    }
    }finally{
        if(curs!=null)
            curs.close();
    }
    return reslist;
}

private <T> List<T> scan(Endpoint ep,String colName, Class<T> cls, Schema schema, String key) {
    ArrayList<T> reslist = new ArrayList<T>();  


    DBCollection c  = db.getCollection(colName);
    DBCursor curs = null;
    try{
        if(ep==null){
            curs = c.find().batchSize(50).addOption(com.mongodb.Bytes.QUERYOPTION_NOTIMEOUT);
        }else{
            BasicDBObject q = new BasicDBObject();
            q.append(key, ep.getUri().toString());
            curs = c.find(q).batchSize(50).addOption(com.mongodb.Bytes.QUERYOPTION_NOTIMEOUT);
        }

        while(curs.hasNext()){
            DBObject o = curs.next();
            SpecificDatumReader r = new SpecificDatumReader<T>(cls);
            JsonDecoder d;
            try {
                d = DecoderFactory.get().jsonDecoder(schema, o.toString());
                T t =(T) r.read(null, d);
                reslist.add(t);
            } catch (IOException e) {
                log.error("SCAN Exception: {}:{}:{} {}", ep.getUri(),colName, cls, ExceptionHandler.logAndtoString(e,true));
            }
        }
    }finally{
        if(curs!=null)
            curs.close();
    }

    return reslist;
}

/**
 * Create a new Event Reader
 * @param in
 * @throws IOException
 */
@SuppressWarnings("deprecation")
public EventReader(DataInputStream in) throws IOException {
  this.in = in;
  this.version = in.readLine();

  if (!EventWriter.VERSION.equals(version)) {
    throw new IOException("Incompatible event log version: "+version);
  }

  this.schema = Schema.parse(in.readLine());
  this.reader = new SpecificDatumReader(schema);
  this.decoder = new JsonDecoder(schema, in);
}

@SuppressWarnings("deprecation")
protected Decoder createDecoder() throws IOException {
  switch(codecType) {
    case BINARY:
      return new BinaryDecoder(getOrCreateInputStream());
    case JSON:
      return new JsonDecoder(schema, getOrCreateInputStream());
  }
  return null;
}

public static Object createGenericRequestFromRecord(Schema requestSchema, JsonNode record) throws IOException {
  JsonDecoder jsonDecoder = DecoderFactory.get().jsonDecoder(requestSchema, record.toString());
  return new GenericDatumReader<>(requestSchema, requestSchema).read(null, jsonDecoder);
}

public static <T> T createSpecificRequestFromRecord(Class<T> requestClass, JsonNode record) throws IOException {
  JsonDecoder jsonDecoder = DecoderFactory.get().jsonDecoder(getSchema(requestClass), record.toString());
  return new SpecificDatumReader<>(requestClass).read(null, jsonDecoder);
}

public static Object createGenericRequestFromRecord(Schema requestSchema, JsonNode record)
    throws IOException {
  JsonDecoder jsonDecoder = DecoderFactory.get().jsonDecoder(requestSchema, record.toString());
  return new GenericDatumReader<>(requestSchema, requestSchema).read(null, jsonDecoder);
}

public static <T> T createSpecificRequestFromRecord(Class<T> requestClass, JsonNode record)
    throws IOException {
  JsonDecoder jsonDecoder =
      DecoderFactory.get().jsonDecoder(getSchema(requestClass), record.toString());
  return new SpecificDatumReader<>(requestClass).read(null, jsonDecoder);
}

private <T> Iterator<T> scanIterator(final Endpoint ep,final String colName, final Class<T> cls, final Schema schema, final String key) {
    ArrayList<T> reslist = new ArrayList<T>();  


    DBCollection c  = db.getCollection(colName);
    final DBCursor curs;
    try{
        if(ep==null){
            curs = c.find().batchSize(50).addOption(com.mongodb.Bytes.QUERYOPTION_NOTIMEOUT);
        }else{
            BasicDBObject q = new BasicDBObject();
            q.append(key, ep.getUri().toString());
            curs = c.find(q).batchSize(50).addOption(com.mongodb.Bytes.QUERYOPTION_NOTIMEOUT);
        }

        return new Iterator<T>() {

            @Override
            public boolean hasNext() {
                // TODO Auto-generated method stub
                return curs.hasNext();
            }

            @Override
            public T next() {
                DBObject o = curs.next();
                SpecificDatumReader r = new SpecificDatumReader<T>(cls);
                JsonDecoder d;
                try {
                    d = DecoderFactory.get().jsonDecoder(schema, o.toString());
                    T t =(T) r.read(null, d);
                    return t;
                } catch (IOException e) {
                    log.error("SCAN Exception: {}:{}:{} {}", ep.getUri(),colName, cls, ExceptionHandler.logAndtoString(e,true));
                }
                return null;
            }

            @Override
            public void remove() {
                // TODO Auto-generated method stub

            }
        };

    }finally{
        //if(curs!=null)
        //  curs.close();
    }


}