Java源码示例：org.apache.flume.serialization.EventSerializer
示例1
@Test
public void testWithNewline() throws FileNotFoundException, IOException {
Map<String, String> headers = new HashMap<String, String>();
headers.put("message", "message1");
OutputStream out = new FileOutputStream(testFile);
CustomLastfmHeaderAndBodyTextEventSerializer.Builder builder = CustomLastfmHeaderAndBodyTextEventSerializer.builder();
EventSerializer serializer = builder.build(new Context(), out);
serializer.afterCreate();
serializer.write(EventBuilder.withBody("messageBody", Charsets.UTF_8, headers));
serializer.flush();
serializer.beforeClose();
out.flush();
out.close();
BufferedReader reader = new BufferedReader(new FileReader(testFile));
Assert.assertEquals("message1", reader.readLine());
Assert.assertNull(reader.readLine());
reader.close();
FileUtils.forceDelete(testFile);
}
示例2
@Test
public void testNoNewline() throws FileNotFoundException, IOException {
Map<String, String> headers = new HashMap<String, String>();
headers.put("header1", "value1");
OutputStream out = new FileOutputStream(testFile);
Context context = new Context();
context.put("appendNewline", "false");
CustomLastfmHeaderAndBodyTextEventSerializer.Builder builder = CustomLastfmHeaderAndBodyTextEventSerializer.builder();
EventSerializer serializer = builder.build(new Context(), out);
serializer.afterCreate();
serializer.write(EventBuilder.withBody("event 1", Charsets.UTF_8, headers));
serializer.write(EventBuilder.withBody("event 2", Charsets.UTF_8, headers));
serializer.write(EventBuilder.withBody("event 3", Charsets.UTF_8, headers));
serializer.flush();
serializer.beforeClose();
out.flush();
out.close();
BufferedReader reader = new BufferedReader(new FileReader(testFile));
Assert.assertNull(reader.readLine());
reader.close();
FileUtils.forceDelete(testFile);
}
示例3
@Override
public void configure(Context context) {
super.configure(context);
serializerType = context.getString("serializer", "TEXT");
useRawLocalFileSystem = context.getBoolean("hdfs.useRawLocalFileSystem",
false);
serializerContext =
new Context(context.getSubProperties(EventSerializer.CTX_PREFIX));
logger.info("Serializer = " + serializerType + ", UseRawLocalFileSystem = "
+ useRawLocalFileSystem);
}
示例4
@Override
public void configure(Context context) {
super.configure(context);
serializerType = context.getString("serializer", "TEXT");
useRawLocalFileSystem = context.getBoolean("hdfs.useRawLocalFileSystem",
false);
serializerContext = new Context(
context.getSubProperties(EventSerializer.CTX_PREFIX));
logger.info("Serializer = " + serializerType + ", UseRawLocalFileSystem = "
+ useRawLocalFileSystem);
}
示例5
@Override
public void configure(Context context) {
super.configure(context);
serializerType = context.getString("serializer", "TEXT");
useRawLocalFileSystem = context.getBoolean("hdfs.useRawLocalFileSystem",
false);
serializerContext =
new Context(context.getSubProperties(EventSerializer.CTX_PREFIX));
logger.info("Serializer = " + serializerType + ", UseRawLocalFileSystem = "
+ useRawLocalFileSystem);
}
示例6
@Override
public void configure(Context context) {
super.configure(context);
serializerType = context.getString("serializer", "TEXT");
useRawLocalFileSystem = context.getBoolean("hdfs.useRawLocalFileSystem",
false);
serializerContext = new Context(
context.getSubProperties(EventSerializer.CTX_PREFIX));
logger.info("Serializer = " + serializerType + ", UseRawLocalFileSystem = "
+ useRawLocalFileSystem);
}
示例7
@Override
public EventSerializer build(Context context, OutputStream out) {
path = context.getString(PATH, PATH_DEFAULT);
customerHeader = context.getString(CUSTOMER_HEADER, CUSTOMER_HEADER_DEFAULT);
hostHeader = context.getString(HOST_HEADER, HOST_HEADER_DEFAULT);
SyslogAvroEventSerializer writer = null;
try {
writer = new SyslogAvroEventSerializer(out, path, customerHeader, hostHeader);
writer.configure(context);
} catch (IOException e) {
log.error("Unable to parse schema file. Exception follows.", e);
}
return writer;
}
示例8
@Override
public EventSerializer build(Context context, OutputStream out) {
ApacheLogAvroEventSerializer writer = null;
try {
writer = new ApacheLogAvroEventSerializer(out);
writer.configure(context);
} catch (IOException e) {
log.error("Unable to parse schema file. Exception follows.", e);
}
return writer;
}
示例9
@Override
public EventSerializer build(Context context, OutputStream out) {
JavaLogAvroEventSerializer writer = null;
try {
writer = new JavaLogAvroEventSerializer(out);
writer.configure(context);
} catch (IOException e) {
log.error("Unable to parse schema file. Exception follows.", e);
}
return writer;
}
示例10
@Override
public EventSerializer build(Context context, OutputStream out) {
AvroEventSerializer writer = new AvroEventSerializer(out);
writer.configure(context);
return writer;
}
示例11
@Override
public EventSerializer build(Context context, OutputStream out) {
return new CustomLastfmHeaderAndBodyTextEventSerializer(out, context);
}
示例12
@Override
public EventSerializer build(Context context, OutputStream out) {
FlumeEventAvroEventSerializer writer = new FlumeEventAvroEventSerializer(out);
writer.configure(context);
return writer;
}
示例13
@Override
public EventSerializer build(Context context, OutputStream out) {
AvroEventSerializer writer = new AvroEventSerializer(out);
writer.configure(context);
return writer;
}
示例14
public void createAvroFile(File file, String codec, boolean useSchemaUrl) throws
IOException {
// serialize a few events using the reflection-based avro serializer
OutputStream out = new FileOutputStream(file);
Context ctx = new Context();
if (codec != null) {
ctx.put("compressionCodec", codec);
}
Schema schema = Schema.createRecord("myrecord", null, null, false);
schema.setFields(Arrays.asList(new Schema.Field[]{
new Schema.Field("message", Schema.create(Schema.Type.STRING), null, null)
}));
GenericRecordBuilder recordBuilder = new GenericRecordBuilder(schema);
File schemaFile = null;
if (useSchemaUrl) {
schemaFile = File.createTempFile(getClass().getSimpleName(), ".avsc");
Files.write(schema.toString(), schemaFile, Charsets.UTF_8);
}
EventSerializer.Builder builder = new AvroEventSerializer.Builder();
EventSerializer serializer = builder.build(ctx, out);
serializer.afterCreate();
for (int i = 0; i < 3; i++) {
GenericRecord record = recordBuilder.set("message", "Hello " + i).build();
Event event = EventBuilder.withBody(serializeAvro(record, schema));
if (schemaFile == null) {
event.getHeaders().put(AvroEventSerializer.AVRO_SCHEMA_LITERAL_HEADER,
schema.toString());
} else {
event.getHeaders().put(AvroEventSerializer.AVRO_SCHEMA_URL_HEADER,
schemaFile.toURI().toURL().toExternalForm());
}
serializer.write(event);
}
serializer.flush();
serializer.beforeClose();
out.flush();
out.close();
}
示例15
@Override
public EventSerializer build(Context context, OutputStream out) {
FlumeEventStringBodyAvroEventSerializer writer = new FlumeEventStringBodyAvroEventSerializer(out);
writer.configure(context);
return writer;
}
示例16
@Override
public EventSerializer build(Context context, OutputStream out) {
FlumeEventStringAvroEventSerializer writer = new FlumeEventStringAvroEventSerializer(out);
writer.configure(context);
return writer;
}
示例17
@Test
public void test() throws FileNotFoundException, IOException {
// create the file, write some data
OutputStream out = new FileOutputStream(testFile);
String builderName = JavaLogAvroEventSerializer.Builder.class.getName();
Context ctx = new Context();
ctx.put("syncInterval", "4096");
EventSerializer serializer =
EventSerializerFactory.getInstance(builderName, ctx, out);
serializer.afterCreate(); // must call this when a file is newly created
List<Event> events = generateJavaEvents();
for (Event e : events) {
serializer.write(e);
}
serializer.flush();
serializer.beforeClose();
out.flush();
out.close();
// now try to read the file back
DatumReader<GenericRecord> reader = new GenericDatumReader<GenericRecord>();
DataFileReader<GenericRecord> fileReader =
new DataFileReader<GenericRecord>(testFile, reader);
GenericRecord record = new GenericData.Record(fileReader.getSchema());
int numEvents = 0;
while (fileReader.hasNext()) {
fileReader.next(record);
long timestamp = (Long) record.get("timestamp");
String datetime = record.get("datetime").toString();
String classname = record.get("classname").toString();
String message = record.get("message").toString();
System.out.println(classname + ": " + message + " (at " + datetime + ")");
numEvents++;
}
fileReader.close();
Assert.assertEquals("Should have found a total of 4 events", 4, numEvents);
FileUtils.forceDelete(testFile);
}
示例18
@Test
public void test() throws FileNotFoundException, IOException {
// create the file, write some data
OutputStream out = new FileOutputStream(testFile);
String builderName = SyslogAvroEventSerializer.Builder.class.getName();
Context ctx = new Context();
ctx.put("syncInterval", "4096");
ctx.put("path", "src/test/resources/customerToHostsFile.txt");
EventSerializer serializer =
EventSerializerFactory.getInstance(builderName, ctx, out);
serializer.afterCreate(); // must call this when a file is newly created
List<Event> events = generateSyslogEvents();
for (Event e : events) {
serializer.write(e);
}
serializer.flush();
serializer.beforeClose();
out.flush();
out.close();
// now try to read the file back
DatumReader<GenericRecord> reader = new GenericDatumReader<GenericRecord>();
DataFileReader<GenericRecord> fileReader =
new DataFileReader<GenericRecord>(testFile, reader);
GenericRecord record = new GenericData.Record(fileReader.getSchema());
int numEvents = 0;
while (fileReader.hasNext()) {
fileReader.next(record);
long timestamp = (Long) record.get("timestamp");
String datetime = record.get("datetime").toString();
String hostname = record.get("hostname").toString();
Map<String, String> headers = (Map<String, String>) record.get("headers");
String message = record.get("message").toString();
System.out.println(hostname + " (" + headers + ")" + ": " + message);
numEvents++;
}
fileReader.close();
Assert.assertEquals("Should have found a total of 6 events", 6, numEvents);
FileUtils.forceDelete(testFile);
}
示例19
@Test
public void test() throws FileNotFoundException, IOException {
// create the file, write some data
OutputStream out = new FileOutputStream(testFile);
String builderName = ApacheLogAvroEventSerializer.Builder.class.getName();
Context ctx = new Context();
ctx.put("syncInterval", "4096");
EventSerializer serializer =
EventSerializerFactory.getInstance(builderName, ctx, out);
serializer.afterCreate(); // must call this when a file is newly created
List<Event> events = generateApacheEvents();
for (Event e : events) {
serializer.write(e);
}
serializer.flush();
serializer.beforeClose();
out.flush();
out.close();
// now try to read the file back
DatumReader<GenericRecord> reader = new GenericDatumReader<GenericRecord>();
DataFileReader<GenericRecord> fileReader =
new DataFileReader<GenericRecord>(testFile, reader);
GenericRecord record = new GenericData.Record(fileReader.getSchema());
int numEvents = 0;
while (fileReader.hasNext()) {
fileReader.next(record);
String ip = record.get("ip").toString();
String uri = record.get("uri").toString();
Integer statuscode = (Integer) record.get("statuscode");
String original = record.get("original").toString();
String connectionstatus = record.get("connectionstatus").toString();
Assert.assertEquals("Ip should be 80.79.194.3", "80.79.194.3", ip);
System.out.println("IP " + ip + " requested: " + uri + " with status code " + statuscode + " and connectionstatus: " + connectionstatus);
System.out.println("Original logline: " + original);
numEvents++;
}
fileReader.close();
Assert.assertEquals("Should have found a total of 3 events", 2, numEvents);
FileUtils.forceDelete(testFile);
}
