org.kitesdk.morphline.base.Notifications.notifyBeginTransaction()方法的使用及代码示例

x33g5p2x  于2022-01-25 转载在 其他  
字(7.7k)|赞(0)|评价(0)|浏览(161)

本文整理了Java中org.kitesdk.morphline.base.Notifications.notifyBeginTransaction()方法的一些代码示例，展示了Notifications.notifyBeginTransaction()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Notifications.notifyBeginTransaction()方法的具体详情如下：
包路径：org.kitesdk.morphline.base.Notifications
类名称：Notifications
方法名：notifyBeginTransaction

Notifications.notifyBeginTransaction介绍

[英]Notify the given command that a transaction has begun.
[中]通知给定命令事务已开始。

代码示例

代码示例来源：origin: apache/flume

@Override
public void beginTransaction() {
 Notifications.notifyBeginTransaction(morphline);      
}

代码示例来源：origin: org.apache.flume.flume-ng-sinks/flume-ng-morphline-solr-sink

@Override
public void beginTransaction() {
 Notifications.notifyBeginTransaction(morphline);      
}

代码示例来源：origin: com.cloudera.search/search-mr

Notifications.notifyBeginTransaction(morphline);

代码示例来源：origin: cloudera/search

Notifications.notifyBeginTransaction(morphline);

代码示例来源：origin: NGDATA/hbase-indexer

MetricRegistry.name("morphline.app", "numExceptionRecords"));
Notifications.notifyBeginTransaction(morphline);

代码示例来源：origin: com.ngdata/hbase-indexer-morphlines

MetricRegistry.name("morphline.app", "numExceptionRecords"));
Notifications.notifyBeginTransaction(morphline);

代码示例来源：origin: cloudera/search

MetricRegistry.name(Metrics.MORPHLINE_APP, Metrics.NUM_EXCEPTION_RECORDS));
Notifications.notifyBeginTransaction(morphline);

代码示例来源：origin: cloudera-labs/envelope

Notifications.notifyBeginTransaction(pipeline.getMorphline());

代码示例来源：origin: kite-sdk/kite

/** Usage: java ... <morphline.conf> <dataFile1> ... <dataFileN> */
 public static void main(String[] args) throws IOException {
  // compile morphline.conf file on the fly
  File morphlineFile = new File(args[0]);
  String morphlineId = null;
  MorphlineContext morphlineContext = new MorphlineContext.Builder().build();
  Command morphline = new Compiler().compile(morphlineFile, morphlineId, morphlineContext, null);
  
  // process each input data file
  Notifications.notifyBeginTransaction(morphline);
  try {
   for (int i = 1; i < args.length; i++) {
    InputStream in = new BufferedInputStream(new FileInputStream(new File(args[i])));
    Record record = new Record();
    record.put(Fields.ATTACHMENT_BODY, in);      
    Notifications.notifyStartSession(morphline);
    boolean success = morphline.process(record);
    if (!success) {
     System.out.println("Morphline failed to process record: " + record);
    }
    in.close();
   }
   Notifications.notifyCommitTransaction(morphline);
  } catch (RuntimeException e) {
   Notifications.notifyRollbackTransaction(morphline);
   morphlineContext.getExceptionHandler().handleException(e, null);
  }
  Notifications.notifyShutdown(morphline);
 }
}

代码示例来源：origin: kite-sdk/kite

@Test
public void testNotifications() throws Exception {
 morphline = createMorphline("test-morphlines/pipeWithTwoBasicCommands");
 Notifications.notifyBeginTransaction(morphline);
 Notifications.notifyStartSession(morphline);
 Notifications.notifyCommitTransaction(morphline);
 Notifications.notifyRollbackTransaction(morphline);
}

代码示例来源：origin: kite-sdk/kite

collector.reset();
startSession();
Notifications.notifyBeginTransaction(morphline);
assertTrue(morphline.process(record));
assertEquals(1, collector.getNumStartEvents());

代码示例来源：origin: kite-sdk/kite

private void runTweetContainer(String morphlineConfigFile, String[] fieldNames) throws Exception {
 File file = new File(RESOURCES_DIR + "/test-documents/sample-statuses-20120906-141433-medium.avro");
 morphline = createMorphline(morphlineConfigFile);    
 for (int j = 0; j < 3; j++) { // also test reuse of objects and low level avro buffers
  Record record = new Record();
  byte[] body = Files.toByteArray(file);    
  record.put(Fields.ATTACHMENT_BODY, body);
  collector.reset();
  startSession();
  Notifications.notifyBeginTransaction(morphline);
  assertTrue(morphline.process(record));
  assertEquals(1, collector.getNumStartEvents());
  assertEquals(2104, collector.getRecords().size());
  
  FileReader<GenericData.Record> reader = new DataFileReader(file, new GenericDatumReader());
  int i = 0;
  while (reader.hasNext()) {
   Record actual = collector.getRecords().get(i);
   GenericData.Record expected = reader.next();
   assertTweetEquals(expected, actual, fieldNames, i);
   i++;
  }    
  assertEquals(collector.getRecords().size(), i);
 }
}

代码示例来源：origin: kite-sdk/kite

record.put(Fields.ATTACHMENT_BODY, body);
startSession();
Notifications.notifyBeginTransaction(morphline);
assertTrue(morphline.process(record));
assertEquals(1, collector.getNumStartEvents());

代码示例来源：origin: kite-sdk/kite

@Test
public void testTokenizeText() throws Exception {
 morphline = createMorphline("test-morphlines" + File.separator + "tokenizeText");
 for (int i = 0; i < 3; i++) {
  Record record = new Record();
  record.put(Fields.MESSAGE, "Hello World!");
  record.put(Fields.MESSAGE, "\nFoo@Bar.com #%()123");
  Record expected = record.copy();
  expected.getFields().putAll("tokens", Arrays.asList("hello", "world", "foo", "bar.com", "123"));
  collector.reset();
  startSession();
  Notifications.notifyBeginTransaction(morphline);
  assertTrue(morphline.process(record));
  assertEquals(1, collector.getNumStartEvents());
  Notifications.notifyCommitTransaction(morphline);
  assertEquals(expected, collector.getFirstRecord());
 }
}

代码示例来源：origin: kite-sdk/kite

@Test
public void testLoadSolrBasic() throws Exception {
 //System.setProperty("ENV_SOLR_HOME", testSolrHome + File.separator + "collection1");
 morphline = createMorphline("test-morphlines" + File.separator + "loadSolrBasic");    
 //System.clearProperty("ENV_SOLR_HOME");
 Record record = new Record();
 record.put(Fields.ID, "id0");
 record.put("first_name", "Nadja"); // will be sanitized
 startSession();
 Notifications.notifyBeginTransaction(morphline);
 assertTrue(morphline.process(record));
 assertEquals(1, collector.getNumStartEvents());
 Notifications.notifyCommitTransaction(morphline);
 Record expected = new Record();
 expected.put(Fields.ID, "id0");
 assertEquals(Arrays.asList(expected), collector.getRecords());
 assertEquals(1, queryResultSetSize("*:*"));
 Notifications.notifyRollbackTransaction(morphline);
 Notifications.notifyShutdown(morphline);
}

代码示例来源：origin: kite-sdk/kite

@Test
public void testLoadSolrWithChildDocuments() throws Exception {
 morphline = createMorphline("test-morphlines" + File.separator + "loadSolrWithChildDocuments");    
 Record record = new Record();
 record.put(Fields.ID, "id0");
 startSession();
 Notifications.notifyBeginTransaction(morphline);
 assertTrue(morphline.process(record));
 assertEquals(1, collector.getNumStartEvents());
 Notifications.notifyCommitTransaction(morphline);
 
 // This parent block join returns the parent records for records 
 // where the child documents contain "bar" in the id field.
 SolrDocumentList docs = query("{!parent which='content_type:parent'}id:bar").getResults();
 assertEquals(1, docs.size());
 assertEquals("id0", docs.get(0).getFirstValue(Fields.ID));
 
 docs = query("*:*").getResults();
 assertEquals(3, docs.size());
}

代码示例来源：origin: kite-sdk/kite

startSession();
assertEquals(1, collector.getNumStartEvents());
Notifications.notifyBeginTransaction(morphline);
assertTrue(morphline.process(record));

代码示例来源：origin: kite-sdk/kite

record.put("user_friends_count", 123);
record.put("text", "myText");
Notifications.notifyBeginTransaction(morphline);
assertTrue(morphline.process(record));
assertEquals(1, collector.getRecords().size());

代码示例来源：origin: kite-sdk/kite

startSession();
assertEquals(1, collector.getNumStartEvents());
Notifications.notifyBeginTransaction(morphline);
assertTrue(morphline.process(record));

代码示例来源：origin: kite-sdk/kite

record.replaceValues(Fields.ID, "id0");
record.replaceValues("first_name", "Nadja"); // will be sanitized
Notifications.notifyBeginTransaction(morphline);
assertTrue(morphline.process(record.copy()));
assertEquals(1, query("*:*").getResults().size());

内容来源于网络，如有侵权，请联系作者删除！

Notifications

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com