import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
public void run(@DataIn(name="source.events", type=StandardEvent.class) View<StandardEvent> input,
                @DataOut(name="target.events", type=StandardEvent.class) View<StandardEvent> output) {

  DatasetReader<StandardEvent> reader = input.newReader();
  DatasetWriter<StandardEvent> writer = output.newWriter();

  try {
    while (reader.hasNext()) {

      writer.write(reader.next());
    }
  } finally {

    Closeables.closeQuietly(reader);
    Closeables.closeQuietly(writer);
  }
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
public void run(@DataIn(name="source_users") View<GenericRecord> input,
                @DataOut(name="target_users") View<GenericRecord> output) {

  DatasetReader<GenericRecord> reader = input.newReader();
  DatasetWriter<GenericRecord> writer = output.newWriter();

  try {
    while (reader.hasNext()) {

      writer.write(reader.next());
    }
  } finally {

    Closeables.closeQuietly(reader);
    Closeables.closeQuietly(writer);
  }
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
public void run(@DataOut(name="example_events", type=ExampleEvent.class) Signalable<ExampleEvent> view) {

    // Write some test data to the view.
    DatasetWriter<ExampleEvent> writer = view.newWriter();

    try {

      for (int i = 0; i < 10; ++i) {

        ExampleEvent event = ExampleEvent.newBuilder()
            .setUserId(i)
            .setSessionId(Integer.toString(i))
            .setTimestamp(getNominalTime().getMillis())
            .build();

        writer.write(event);
      }
    }
    finally {
      writer.close();
    }

    // Signal that our view is ready.
    view.signalReady();
  }
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
@Override
public int run(List<String> args) throws Exception {

  Preconditions.checkState(!Datasets.exists(uri),
      "events dataset already exists");

  DatasetDescriptor descriptor = new DatasetDescriptor.Builder()
      .schema(StandardEvent.class).build();

  View<StandardEvent> events = Datasets.create(uri, descriptor, StandardEvent.class);
  DatasetWriter<StandardEvent> writer = events.newWriter();
  try {
    while (System.currentTimeMillis() - baseTimestamp < 36000) {
      writer.write(generateRandomEvent());
    }
  } finally {
    writer.close();
  }

  System.out.println("Generated " + counter + " events");

  return 0;
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
@Test
public void testAppendWriteExceptionInvokesPolicy()
    throws EventDeliveryException, NonRecoverableEventException {
  DatasetSink sink = sink(in, config);

  // run the sink
  sink.start();
  sink.process();

  // Mock an Event
  Event mockEvent = mock(Event.class);
  when(mockEvent.getBody()).thenReturn(new byte[] { 0x01 });

  // Mock a GenericRecord
  GenericRecord mockRecord = mock(GenericRecord.class);

  // Mock an EntityParser
  EntityParser<GenericRecord> mockParser = mock(EntityParser.class);
  when(mockParser.parse(eq(mockEvent), any(GenericRecord.class)))
      .thenReturn(mockRecord);
  sink.setParser(mockParser);

  // Mock a FailurePolicy
  FailurePolicy mockFailurePolicy = mock(FailurePolicy.class);
  sink.setFailurePolicy(mockFailurePolicy);

  // Mock a DatasetWriter
  DatasetWriter<GenericRecord> mockWriter = mock(DatasetWriter.class);
  doThrow(new DataFileWriter.AppendWriteException(new IOException()))
      .when(mockWriter).write(mockRecord);

  sink.setWriter(mockWriter);
  sink.write(mockEvent);

  // Verify that the event was sent to the failure policy
  verify(mockFailurePolicy).handle(eq(mockEvent), any(Throwable.class));

  sink.stop();
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
private DatasetWriter<GenericRecord> newWriter(
    final UserGroupInformation login, final URI uri) {
  View<GenericRecord> view = KerberosUtil.runPrivileged(login,
      new PrivilegedExceptionAction<Dataset<GenericRecord>>() {
        @Override
        public Dataset<GenericRecord> run() {
          return Datasets.load(uri);
        }
      });

  DatasetDescriptor descriptor = view.getDataset().getDescriptor();
  String formatName = descriptor.getFormat().getName();
  Preconditions.checkArgument(allowedFormats().contains(formatName),
      "Unsupported format: " + formatName);

  Schema newSchema = descriptor.getSchema();
  if (targetSchema == null || !newSchema.equals(targetSchema)) {
    this.targetSchema = descriptor.getSchema();
    // target dataset schema has changed, invalidate all readers based on it
    readers.invalidateAll();
  }

  this.reuseDatum = !("parquet".equals(formatName));
  this.datasetName = view.getDataset().getName();

  return view.newWriter();
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
@Override
public int run(String[] args) throws Exception {
  // Create a partition strategy that hash partitions on username with 10 buckets
  PartitionStrategy partitionStrategy = new PartitionStrategy.Builder()
      .identity("favoriteColor", "favorite_color")
      .build();

  // Create a dataset of users with the Avro schema
  DatasetDescriptor descriptor = new DatasetDescriptor.Builder()
      .schemaUri("resource:user.avsc")
      .partitionStrategy(partitionStrategy)
      .build();
  Dataset<Record> users = Datasets.create(
      "dataset:hdfs:/tmp/data/users", descriptor, Record.class);

  // Get a writer for the dataset and write some users to it
  DatasetWriter<Record> writer = null;
  try {
    writer = users.newWriter();
    Random rand = new Random();
    GenericRecordBuilder builder = new GenericRecordBuilder(descriptor.getSchema());
    for (int i = 0; i < 100; i++) {
      Record record = builder.set("username", "user-" + i)
          .set("creationDate", System.currentTimeMillis())
          .set("favoriteColor", colors[rand.nextInt(colors.length)]).build();
      writer.write(record);
    }
  } finally {
    if (writer != null) {
      writer.close();
    }
  }

  return 0;
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
@Override
public int run(String[] args) throws Exception {
  // Create a dataset of users with the Avro schema
  DatasetDescriptor descriptor = new DatasetDescriptor.Builder()
      .schemaUri("resource:user.avsc")
      .build();
  Dataset<Record> users = Datasets.create(
      "dataset:hdfs:/tmp/data/users", descriptor, Record.class);

  // Get a writer for the dataset and write some users to it
  DatasetWriter<Record> writer = null;
  try {
    writer = users.newWriter();
    Random rand = new Random();
    GenericRecordBuilder builder = new GenericRecordBuilder(descriptor.getSchema());
    for (int i = 0; i < 100; i++) {
      Record record = builder.set("username", "user-" + i)
          .set("creationDate", System.currentTimeMillis())
          .set("favoriteColor", colors[rand.nextInt(colors.length)]).build();
      writer.write(record);
    }
  } finally {
    if (writer != null) {
      writer.close();
    }
  }

  return 0;
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
@Override
public int run(String[] args) throws Exception {
  DatasetDescriptor descriptor = new DatasetDescriptor.Builder()
      .schemaUri("resource:user.avsc")
      .format(Formats.PARQUET)
      .build();
  Dataset<Record> users = Datasets.create(
      "dataset:hdfs:/tmp/data/users", descriptor, Record.class);

  // Get a writer for the dataset and write some users to it
  DatasetWriter<Record> writer = null;
  try {
    writer = users.newWriter();
    Random rand = new Random();
    GenericRecordBuilder builder = new GenericRecordBuilder(descriptor.getSchema());
    for (int i = 0; i < 100; i++) {
      Record record = builder.set("username", "user-" + i)
          .set("creationDate", System.currentTimeMillis())
          .set("favoriteColor", colors[rand.nextInt(colors.length)]).build();
      writer.write(record);
    }
  } finally {
    if (writer != null) {
      writer.close();
    }
  }

  return 0;
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
@Override
public int run(String[] args) throws Exception {

  // Create a dataset of products with the Avro schema
  DatasetDescriptor descriptor = new DatasetDescriptor.Builder()
      .schema(Product.class)
      .build();
  Dataset<Product> products = Datasets.create(
      "dataset:hdfs:/tmp/data/products", descriptor, Product.class);

  // Get a writer for the dataset and write some products to it
  DatasetWriter<Product> writer = null;
  try {
    writer = products.newWriter();
    int i = 0;
    for (String name : names) {
      Product product = new Product();
      product.setName(name);
      product.setId(i++);
      writer.write(product);
    }
  } finally {
    if (writer != null) {
      writer.close();
    }
  }

  return 0;
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
@Override
public int run(String[] args) throws Exception {
  // Create a dataset of users with the Avro schema
  DatasetDescriptor descriptor = new DatasetDescriptor.Builder()
      .schemaUri("resource:user.avsc")
      .build();
  Dataset<Record> users = Datasets.create("dataset:hive?dataset=users",
      descriptor, Record.class);

  // Get a writer for the dataset and write some users to it
  DatasetWriter<Record> writer = null;
  try {
    writer = users.newWriter();
    Random rand = new Random();
    GenericRecordBuilder builder = new GenericRecordBuilder(descriptor.getSchema());
    for (int i = 0; i < 100; i++) {
      Record record = builder.set("username", "user-" + i)
          .set("creationDate", System.currentTimeMillis())
          .set("favoriteColor", colors[rand.nextInt(colors.length)]).build();
      writer.write(record);
    }

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

  return 0;
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
@VisibleForTesting
DatasetWriter<GenericRecord> getWriter() {
  return writer;
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
@VisibleForTesting
void setWriter(DatasetWriter<GenericRecord> writer) {
  this.writer = writer;
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
@Test
public void testRuntimeExceptionThrowsEventDeliveryException()
    throws EventDeliveryException, NonRecoverableEventException {
  DatasetSink sink = sink(in, config);

  // run the sink
  sink.start();
  sink.process();

  // Mock an Event
  Event mockEvent = mock(Event.class);
  when(mockEvent.getBody()).thenReturn(new byte[] { 0x01 });

  // Mock a GenericRecord
  GenericRecord mockRecord = mock(GenericRecord.class);

  // Mock an EntityParser
  EntityParser<GenericRecord> mockParser = mock(EntityParser.class);
  when(mockParser.parse(eq(mockEvent), any(GenericRecord.class)))
      .thenReturn(mockRecord);
  sink.setParser(mockParser);

  // Mock a FailurePolicy
  FailurePolicy mockFailurePolicy = mock(FailurePolicy.class);
  sink.setFailurePolicy(mockFailurePolicy);

  // Mock a DatasetWriter
  DatasetWriter<GenericRecord> mockWriter = mock(DatasetWriter.class);
  doThrow(new RuntimeException()).when(mockWriter).write(mockRecord);

  sink.setWriter(mockWriter);

  try {
    sink.write(mockEvent);
    Assert.fail("Should throw EventDeliveryException");
  } catch (EventDeliveryException ex) {

  }

  // Verify that the event was not sent to the failure policy
  verify(mockFailurePolicy, never()).handle(eq(mockEvent), any(Throwable.class));

  sink.stop();
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
private DatasetWriter<GenericRecord> getOrNewWriter() {
  if (writer == null) {
    writer = dataset.newWriter();
  }
  return writer;
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
public void run(@DataOut(name="kv-output", type= KeyValues.class) View<KeyValues> output) {


    DatasetWriter<KeyValues> writer = output.newWriter();

    try {

      JobContext context = getJobContext();

      KeyValues kv = KeyValues.newBuilder()
          .setJobsettings(context.getSettings())
          .setOutputsettings(context.getOutputSettings("kv-output"))
          .build();

      writer.write(kv);

    } finally {


      Closeables.closeQuietly(writer);
    }
  }
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
@Override
public int run(String[] args) throws Exception {
  // going to generate a lot of random log messages
  final Random rand = new Random();

  // data is written to the staging dataset
  Dataset<Record> staging = Datasets.load(
      "dataset:file:/tmp/data/logs_staging", Record.class);

  // this is going to build our simple log records
  GenericRecordBuilder builder = new GenericRecordBuilder(
      staging.getDescriptor().getSchema());

  // generate timestamps 1 second apart starting 1 day ago
  final Calendar now = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
  final long yesterday = now.getTimeInMillis() - DAY_IN_MILLIS;

  DatasetWriter<Record> writer = null;
  try {
    writer = staging.newWriter();

    // generate 15,000 messages, each 5 seconds apart, starting 24 hours ago
    // this is a little less than 24 hours worth of messages
    for (int second : Ranges.closed(0, 15000).asSet(DiscreteDomains.integers())) {
      LOG.info("Generating log message " + second);

      builder.set("timestamp", yesterday + second * 5000);
      builder.set("component", "GenerateSimpleLogs");

      int level = rand.nextInt(LOG_LEVELS.length);
      builder.set("level", LOG_LEVELS[level]);
      builder.set("message", LOG_MESSAGES[level]);

      writer.write(builder.build());
    }

    if (writer instanceof Flushable) {
      ((Flushable) writer).flush();
    }
  } finally {
    if (writer != null) {
      writer.close();
    }
  }

  return 0;
}
 

import org.kitesdk.data.DatasetWriter; //导入依赖的package包/类
@Override
public void runJob(Map params, JobContext jobContext, Instant nominalTime) {

  View output = (View) params.get("output");

  DatasetWriter<GenericRecord> writer = output.newWriter();

  // Simply write all of the input datasets to the output.
  for (Map.Entry<String,View> param: ((Map<String,View>)  params).entrySet()) {

    if (!param.getKey().equals("output")) {
      DatasetReader<GenericRecord> reader = param.getValue().newReader();

      try {
        while (reader.hasNext()) {

          writer.write(reader.next());
        }
      } finally {

        Closeables.closeQuietly(reader);

      }
    }
  }

  Closeables.closeQuietly(writer);
}