import org.kitesdk.data.Datasets; //导入依赖的package包/类
private SavePolicy(Context context) {
  String uri = context.getString(CONFIG_KITE_ERROR_DATASET_URI);
  Preconditions.checkArgument(uri != null, "Must set "
      + CONFIG_KITE_ERROR_DATASET_URI + " when " + CONFIG_FAILURE_POLICY
      + "=save");
  if (Datasets.exists(uri)) {
    dataset = Datasets.load(uri, AvroFlumeEvent.class);
  } else {
    DatasetDescriptor descriptor = new DatasetDescriptor.Builder()
        .schema(AvroFlumeEvent.class)
        .build();
    dataset = Datasets.create(uri, descriptor, AvroFlumeEvent.class);
  }

  nEventsHandled = 0;
}
 

import org.kitesdk.data.Datasets; //导入依赖的package包/类
@Test
public void testOldConfig() throws EventDeliveryException {
  config.put(DatasetSinkConstants.CONFIG_KITE_DATASET_URI, null);
  config.put(DatasetSinkConstants.CONFIG_KITE_REPO_URI, FILE_REPO_URI);
  config.put(DatasetSinkConstants.CONFIG_KITE_DATASET_NAME, DATASET_NAME);

  DatasetSink sink = sink(in, config);

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

  Assert.assertEquals(
      Sets.newHashSet(expected),
      read(Datasets.load(FILE_DATASET_URI)));
  Assert.assertEquals("Should have committed", 0, remaining(in));
}
 

import org.kitesdk.data.Datasets; //导入依赖的package包/类
@Test
public void testDatasetUriOverridesOldConfig() throws EventDeliveryException {
  // CONFIG_KITE_DATASET_URI is still set, otherwise this will cause an error
  config.put(DatasetSinkConstants.CONFIG_KITE_REPO_URI, "bad uri");
  config.put(DatasetSinkConstants.CONFIG_KITE_DATASET_NAME, "");

  DatasetSink sink = sink(in, config);

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

  Assert.assertEquals(
      Sets.newHashSet(expected),
      read(Datasets.load(FILE_DATASET_URI)));
  Assert.assertEquals("Should have committed", 0, remaining(in));
}
 

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

  // release one record per process call
  config.put("kite.batchSize", "2");
  Configurables.configure(sink, config);

  sink.start();
  sink.process(); // process the first and second
  sink.roll(); // roll at the next process call
  sink.process(); // roll and process the third
  Assert.assertEquals(
      Sets.newHashSet(expected.subList(0, 2)),
      read(Datasets.load(FILE_DATASET_URI)));
  Assert.assertEquals("Should have committed", 0, remaining(in));
  sink.roll(); // roll at the next process call
  sink.process(); // roll, the channel is empty
  Assert.assertEquals(
      Sets.newHashSet(expected),
      read(Datasets.load(FILE_DATASET_URI)));
  sink.stop();
}
 

import org.kitesdk.data.Datasets; //导入依赖的package包/类
@Test
public void testTimedFileRolling()
    throws EventDeliveryException, InterruptedException {
  // use a new roll interval
  config.put("kite.rollInterval", "1"); // in seconds

  DatasetSink sink = sink(in, config);

  Dataset<GenericRecord> records = Datasets.load(FILE_DATASET_URI);

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

  Assert.assertEquals("Should have committed", 0, remaining(in));

  Thread.sleep(1100); // sleep longer than the roll interval
  sink.process(); // rolling happens in the process method

  Assert.assertEquals(Sets.newHashSet(expected), read(records));

  // wait until the end to stop because it would close the files
  sink.stop();
}
 

import org.kitesdk.data.Datasets; //导入依赖的package包/类
@Test
public void testFileStoreWithSavePolicy() throws EventDeliveryException {
  if (Datasets.exists(ERROR_DATASET_URI)) {
    Datasets.delete(ERROR_DATASET_URI);
  }
  config.put(DatasetSinkConstants.CONFIG_FAILURE_POLICY,
      DatasetSinkConstants.SAVE_FAILURE_POLICY);
  config.put(DatasetSinkConstants.CONFIG_KITE_ERROR_DATASET_URI,
      ERROR_DATASET_URI);
  DatasetSink sink = sink(in, config);

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

  Assert.assertEquals(
      Sets.newHashSet(expected),
      read(Datasets.load(FILE_DATASET_URI)));
  Assert.assertEquals("Should have committed", 0, remaining(in));
}
 

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

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

  // the transaction should commit during the call to process
  Assert.assertEquals("Should have committed", 0, remaining(in));
  // but the data won't be visible yet
  Assert.assertEquals(0,
      read(Datasets.load(FILE_DATASET_URI)).size());

  sink.stop();

  Assert.assertEquals(
      Sets.newHashSet(expected),
      read(Datasets.load(FILE_DATASET_URI)));
}
 

import org.kitesdk.data.Datasets; //导入依赖的package包/类
@Test
public void testCloseAndCreateWriter() throws EventDeliveryException {
  config.put(DatasetSinkConstants.CONFIG_FLUSHABLE_COMMIT_ON_BATCH,
      Boolean.toString(false));
  config.put(DatasetSinkConstants.CONFIG_SYNCABLE_SYNC_ON_BATCH,
      Boolean.toString(false));
  DatasetSink sink = sink(in, config);

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

  sink.closeWriter();
  sink.commitTransaction();
  sink.createWriter();

  Assert.assertNotNull("Writer should not be null", sink.getWriter());
  Assert.assertEquals("Should have committed", 0, remaining(in));

  sink.stop();

  Assert.assertEquals(
      Sets.newHashSet(expected),
      read(Datasets.load(FILE_DATASET_URI)));
}
 

import org.kitesdk.data.Datasets; //导入依赖的package包/类
@Test
public void testCloseWriter() throws EventDeliveryException {
  config.put(DatasetSinkConstants.CONFIG_FLUSHABLE_COMMIT_ON_BATCH,
      Boolean.toString(false));
  config.put(DatasetSinkConstants.CONFIG_SYNCABLE_SYNC_ON_BATCH,
      Boolean.toString(false));
  DatasetSink sink = sink(in, config);

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

  sink.closeWriter();
  sink.commitTransaction();

  Assert.assertNull("Writer should be null", sink.getWriter());
  Assert.assertEquals("Should have committed", 0, remaining(in));

  sink.stop();

  Assert.assertEquals(
      Sets.newHashSet(expected),
      read(Datasets.load(FILE_DATASET_URI)));
}
 

import org.kitesdk.data.Datasets; //导入依赖的package包/类
@Test
public void testCreateWriter() throws EventDeliveryException {
  config.put(DatasetSinkConstants.CONFIG_FLUSHABLE_COMMIT_ON_BATCH,
      Boolean.toString(false));
  config.put(DatasetSinkConstants.CONFIG_SYNCABLE_SYNC_ON_BATCH,
      Boolean.toString(false));
  DatasetSink sink = sink(in, config);

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

  sink.commitTransaction();
  sink.createWriter();
  Assert.assertNotNull("Writer should not be null", sink.getWriter());
  Assert.assertEquals("Should have committed", 0, remaining(in));

  sink.stop();

  Assert.assertEquals(0, read(Datasets.load(FILE_DATASET_URI)).size());
}
 

import org.kitesdk.data.Datasets; //导入依赖的package包/类
@BeforeMethod(alwaysRun = true)
public void setUp() {
  initMocks(this);
  mockStatic(KiteDatasetExecutor.class);
  mockStatic(Datasets.class);

  destroyer = new KiteToDestroyer() {
    @Override
    protected KiteDatasetExecutor getExecutor(String uri, Schema schema,
        FileFormat format) {
      return executorMock;
    }
  };

  linkConfig = new LinkConfiguration();
  toJobConfig = new ToJobConfiguration();
  toJobConfig.toJobConfig.uri = "dataset:file:/foo/bar";
  toJobConfig.toJobConfig.fileFormat = FileFormat.AVRO;
}
 

import org.kitesdk.data.Datasets; //导入依赖的package包/类
/**
 * Ensures the given dataset exists, creating it if it doesn't
 * and updating the schema if necessary.
 */
protected void dataset(String uri, DatasetDescriptor descriptor) {

  try {

    Datasets.create(uri, descriptor);
  } catch (DatasetExistsException e) {

    Dataset existingDataset = Datasets.load(uri);

    DatasetDescriptor updated;

    // The given discriptor might not have a location,
    // so use the current one.
    if (descriptor.getLocation() == null) {
      updated = new DatasetDescriptor.Builder(descriptor)
          .location(existingDataset.getDescriptor().getLocation())
          .build();
    } else {

      updated = descriptor;
    }

    Datasets.update(uri, updated);
  }
}
 

import org.kitesdk.data.Datasets; //导入依赖的package包/类
@Override
public int run(String[] args) throws Exception {
  // Load the users dataset
  Dataset<Record> users = Datasets.load(
      "dataset:hdfs:/tmp/data/users", Record.class);

  // Get a reader for the dataset and read all the users
  DatasetReader<Record> reader = null;
  try {
    reader = users.newReader();
    for (GenericRecord user : reader) {
      System.out.println(user);
    }

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

  return 0;
}
 

import org.kitesdk.data.Datasets; //导入依赖的package包/类
@Override
public int run(String[] args) throws Exception {
  // Load the users dataset
  Dataset<Record> users = Datasets.load(
      "dataset:hive?dataset=users", Record.class);

  // Get a reader for the dataset and read all the users
  DatasetReader<Record> reader = null;
  try {
    reader = users.newReader();
    for (GenericRecord user : users.newReader()) {
      System.out.println(user);
    }

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

  return 0;
}
 

import org.kitesdk.data.Datasets; //导入依赖的package包/类
@Override
public int run(String[] args) throws Exception {
  // Load the users dataset
  Dataset<Record> users = Datasets.load(
      "dataset:hdfs:/tmp/data/users", Record.class);

  // Get a reader for the dataset and read all the users
  DatasetReader<Record> reader = null;
  try {
    reader = users.with("favoriteColor", "green").newReader();
    for (GenericRecord user : reader) {
      System.out.println(user);
    }

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

  return 0;
}
 

import org.kitesdk.data.Datasets; //导入依赖的package包/类
@Override
public int run(String[] args) throws Exception {
  // Load the products dataset
  Dataset<Product> products = Datasets.load(
      "dataset:hdfs:/tmp/data/products", Product.class);

  // Get a reader for the dataset and read all the users
  DatasetReader<Product> reader = null;
  try {
    reader = products.newReader();
    for (Product product : reader) {
      System.out.println(product);
    }

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

  return 0;
}
 

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

  // Load the events dataset
  Dataset<GenericRecord> events = Datasets.load("dataset:hive:/tmp/data/default/events");

  // Get a reader for the dataset and read all the events
  DatasetReader<GenericRecord> reader = events.newReader();
  try {
    for (GenericRecord event : reader) {
      System.out.println(event);
    }
  } finally {
    reader.close();
  }

  return 0;
}
 

import org.kitesdk.data.Datasets; //导入依赖的package包/类
@Override
public int run(String[] args) throws Exception {
  Dataset<Record> movies = Datasets.load(
      "dataset:hdfs:/tmp/data/movies", Record.class);

  DatasetReader<Record> reader = null;
  try {
    reader = movies.newReader();
    for (Record rec : reader) {
      System.err.println("Movie: " + rec);
    }

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

  return 0;
}
 

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

  String inputUri = uri;
  String outputUri = "dataset:hive?dataset=correlated_events";

  if (args.size() == 1) {
    outputUri = args.get(0);
  }

  Preconditions.checkState(Datasets.exists(inputUri),
      "input dataset doesn't exists");

  if (!Datasets.exists(outputUri)) {
    Datasets.create(outputUri, new DatasetDescriptor.Builder()
        .format("avro")
        .schema(CorrelatedEvents.class)
        .build());
  }
  CorrelateEventsTask task = new CorrelateEventsTask(inputUri, outputUri);
  task.run();

  return 0;
}
 

import org.kitesdk.data.Datasets; //导入依赖的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;
}