org.springframework.boot.actuate.health.HealthIndicator#com.hazelcast.jet.JetInstance源码实例Demo

@Bean
JetInstance jetInstance(HazelcastJetServerProperty serverProperty,
                        HazelcastJetIMDGProperty imdgProperty) throws IOException {
    Resource serverConfigLocation = serverProperty.resolveConfigLocation();
    Resource imdgConfigLocation = imdgProperty.resolveConfigLocation();

    JetConfig jetConfig = serverConfigLocation != null ? getJetConfig(serverConfigLocation)
            : ConfigProvider.locateAndGetJetConfig();
    if (imdgConfigLocation != null) {
        jetConfig.setHazelcastConfig(getIMDGConfig(imdgConfigLocation));
    }

    injectSpringManagedContext(jetConfig.getHazelcastConfig());

    return Jet.newJetInstance(jetConfig);
}
 

public static void main(String[] args) {
    JetInstance jet = JetBootstrap.getInstance();

    Properties properties = new Properties();
    properties.setProperty("group.id", "cdc-demo");
    properties.setProperty("bootstrap.servers", "kafka:9092");
    properties.setProperty("key.deserializer", JsonDeserializer.class.getCanonicalName());
    properties.setProperty("value.deserializer", JsonDeserializer.class.getCanonicalName());
    properties.setProperty("auto.offset.reset", "earliest");
    Pipeline p = Pipeline.create();

    p.readFrom(KafkaSources.kafka(properties, record -> {
        HazelcastJsonValue key = new HazelcastJsonValue(record.key().toString());
        HazelcastJsonValue value = new HazelcastJsonValue(record.value().toString());
        return Util.entry(key, value);
    }, "dbserver1.inventory.customers"))
     .withoutTimestamps()
     .peek()
     .writeTo(Sinks.map("customers"));

    jet.newJob(p).join();
}
 

public static void main(String[] args) {
    if (FlightDataSource.API_AUTHENTICATION_KEY.equals("YOUR_API_KEY_HERE")) {
         System.err.println("API_AUTHENTICATION_KEY not set in FlightDataSource.java");
         System.exit(1);
    }

    JetInstance jet = getJetInstance();

    Pipeline pipeline = buildPipeline();
    addListener(jet.getMap(TAKE_OFF_MAP), a -> System.out.println("New aircraft taking off: " + a));
    addListener(jet.getMap(LANDING_MAP), a -> System.out.println("New aircraft landing " + a));

    try {
        Job job = jet.newJob(pipeline, new JobConfig().setName("FlightTelemetry").setProcessingGuarantee(ProcessingGuarantee.EXACTLY_ONCE));
        job.join();
    } finally {
        Jet.shutdownAll();
    }
}
 

public static void main(String[] args) {
    if (args.length != 2) {
        System.err.println("Missing command-line arguments: <input file> <output directory>");
        System.exit(1);
    }

    Path sourceFile = Paths.get(args[0]).toAbsolutePath();
    final String targetDirectory = args[1];
    if (!Files.isReadable(sourceFile)) {
        System.err.println("Source file does not exist or is not readable (" + sourceFile + ")");
        System.exit(1);
    }

    JetInstance instance = Jet.newJetInstance();
    Pipeline pipeline = buildPipeline(sourceFile, targetDirectory);
    try {
        instance.newJob(pipeline).join();
    } finally {
        Jet.shutdownAll();
    }
}
 

public static void main(String[] args) {
    validateWebcam();
    if (args.length != 1) {
        System.err.println("Missing command-line argument: <model path>");
        System.exit(1);
    }

    Path modelPath = Paths.get(args[0]).toAbsolutePath();
    if (!Files.isDirectory(modelPath)) {
        System.err.println("Model path does not exist (" + modelPath + ")");
        System.exit(1);
    }

    Pipeline pipeline = buildPipeline();

    JobConfig jobConfig = new JobConfig();
    jobConfig.attachDirectory(modelPath.toString(), "model");

    JetInstance jet = Jet.newJetInstance();
    try {
        jet.newJob(pipeline, jobConfig).join();
    } finally {
        Jet.shutdownAll();
    }
}
 

public static void main(String[] args) {
    System.setProperty("hazelcast.logging.type", "log4j");

    if (args.length != 2) {
        System.out.println("Usage: ModelServerClassification <data path> <model server address>");
        System.exit(1);
    }
    String dataPath = args[0];
    String serverAddress = args[1];

    JobConfig jobConfig = new JobConfig();
    jobConfig.attachDirectory(dataPath, "data");

    JetInstance instance = Jet.newJetInstance();
    try {
        IMap<Long, String> reviewsMap = instance.getMap("reviewsMap");
        SampleReviews.populateReviewsMap(reviewsMap);

        Pipeline p = buildPipeline(serverAddress, reviewsMap);

        instance.newJob(p, jobConfig).join();
    } finally {
        instance.shutdown();
    }
}
 

public static void main(String[] args) {
    System.setProperty("hazelcast.logging.type", "log4j");

    if (args.length != 1) {
        System.out.println("Usage: InProcessClassification <data path>");
        System.exit(1);
    }

    String dataPath = args[0];
    JetInstance instance = Jet.newJetInstance();
    JobConfig jobConfig = new JobConfig();
    jobConfig.attachDirectory(dataPath, "data");

    try {
        IMap<Long, String> reviewsMap = instance.getMap("reviewsMap");
        SampleReviews.populateReviewsMap(reviewsMap);
        instance.newJob(buildPipeline(reviewsMap), jobConfig).join();
    } finally {
        instance.shutdown();
    }
}
 

public static void main(String[] args) {
    Pipeline p = buildPipeline();

    JetInstance jet = Jet.bootstrappedInstance();

    try {
        JobConfig jobConfig = new JobConfig()
                .setAutoScaling(true)
                .setName("TradeAnalysis")
                .setProcessingGuarantee(ProcessingGuarantee.EXACTLY_ONCE);

        jet.newJob(p, jobConfig).join();
    } finally {
        Jet.shutdownAll();
    }
}
 

public static void main(String[] args) {
    JetInstance jet = Jet.bootstrappedInstance();

    // symbol -> company name
    // random symbols from https://www.nasdaq.com
    IMap<String, String> lookupTable = jet.getMap(LOOKUP_TABLE);
    lookupTable.put("AAPL", "Apple Inc. - Common Stock");
    lookupTable.put("GOOGL", "Alphabet Inc.");
    lookupTable.put("MSFT", "Microsoft Corporation");

    Pipeline p = buildPipeline(lookupTable);

    try {
        jet.newJob(p).join();
    } finally {
        jet.shutdown();
    }
}
 

public static void main (String[] args) {
    JetInstance jet = Jet.bootstrappedInstance();

    // symbol -> company name
    IMap<String, String> lookupTable = jet.getMap(LOOKUP_TABLE);
    lookupTable.put("AAPL", "Apple Inc. - Common Stock");
    lookupTable.put("GOOGL", "Alphabet Inc.");
    lookupTable.put("MSFT", "Microsoft Corporation");

    Pipeline p = buildPipeline(lookupTable);
    try {
        jet.newJob(p).join();
    } finally {
        jet.shutdown();
    }
}
 

private JetPipelineResult run(DAG dag) {
  startClusterIfNeeded(options);

  JetInstance jet =
      getJetInstance(
          options); // todo: we use single client for each job, it might be better to have a
  // shared client with refcount

  Job job = jet.newJob(dag, getJobConfig(options));
  IMap<String, MetricUpdates> metricsAccumulator =
      jet.getMap(JetMetricsContainer.getMetricsMapName(job.getId()));
  JetPipelineResult pipelineResult = new JetPipelineResult(job, metricsAccumulator);
  CompletableFuture<Void> completionFuture =
      job.getFuture()
          .whenCompleteAsync(
              (r, f) -> {
                pipelineResult.freeze(f);
                metricsAccumulator.destroy();
                jet.shutdown();

                stopClusterIfNeeded(options);
              });
  pipelineResult.setCompletionFuture(completionFuture);

  return pipelineResult;
}
 

public Long countWord(List<String> sentences, String word) {
    long count = 0;
    JetInstance jet = Jet.newJetInstance();
    try {
        List<String> textList = jet.getList(LIST_NAME);
        textList.addAll(sentences);
        Pipeline p = createPipeLine();
        jet.newJob(p)
            .join();
        Map<String, Long> counts = jet.getMap(MAP_NAME);
        count = counts.get(word);
    } finally {
        Jet.shutdownAll();
    }
    return count;
}
 

@Test
public void when_readFromPulsarConsumer_then_jobGetsAllPublishedMessages() {
    JetInstance[] instances = new JetInstance[2];
    Arrays.setAll(instances, i -> createJetMember());

    String topicName = randomName();
    StreamSource<String> pulsarConsumerSrc = setupConsumerSource(topicName,
            x -> new String(x.getData(), StandardCharsets.UTF_8));

    Pipeline pipeline = Pipeline.create();
    pipeline.readFrom(pulsarConsumerSrc)
            .withoutTimestamps()
            .writeTo(AssertionSinks.assertCollectedEventually(60,
                    list -> {
                        assertEquals("# of Emitted items should be equal to # of published items",
                                ITEM_COUNT, list.size());
                        for (int i = 0; i < ITEM_COUNT; i++) {
                            String message = "hello-pulsar-" + i;
                            Assert.assertTrue("missing entry: " + message, list.contains(message));
                        }
                    })
            );
    Job job = instances[0].newJob(pipeline);
    assertJobStatusEventually(job, JobStatus.RUNNING);

    produceMessages("hello-pulsar", topicName, ITEM_COUNT);

    try {
        job.join();
        fail("Job should have completed with an AssertionCompletedException, but completed normally");
    } catch (CompletionException e) {
        String errorMsg = e.getCause().getMessage();
        assertTrue("Job was expected to complete with AssertionCompletedException, but completed with: "
                + e.getCause(), errorMsg.contains(AssertionCompletedException.class.getName()));
    }
    for (JetInstance instance:instances) {
        instance.shutdown();
    }
}
 

@Bean
JetInstance jetInstance(HazelcastJetClientProperty properties) throws IOException {
    Resource configLocation = properties.resolveConfigLocation();
    if (configLocation == null) {
        return Jet.newJetClient();
    }
    return Jet.newJetClient(getClientConfig(configLocation));
}
 

@Test
public void whenAutoConfigured_thenHazelcastJetUp() {
    this.contextRunner.run((context) -> {
        assertThat(context).hasSingleBean(JetInstance.class).hasSingleBean(HazelcastJetHealthIndicator.class);
        JetInstance jet = context.getBean(JetInstance.class);
        Health health = context.getBean(HazelcastJetHealthIndicator.class).health();
        assertThat(health.getStatus()).isEqualTo(Status.UP);
        assertThat(health.getDetails())
                .containsOnlyKeys("name", "uuid")
                .containsEntry("name", jet.getName())
                .containsEntry("uuid", jet.getHazelcastInstance().getLocalEndpoint().getUuid().toString());
    });
}
 

@Test
public void whenShutdown_thenHazelcastJetDown() {
    this.contextRunner.run((context) -> {
        context.getBean(JetInstance.class).shutdown();
        assertThat(context).hasSingleBean(HazelcastJetHealthIndicator.class);
        Health health = context.getBean(HazelcastJetHealthIndicator.class).health();
        assertThat(health.getStatus()).isEqualTo(Status.DOWN);
    });
}
 

@Test
public void defaultConfigFile() {
    // hazelcast-jet.yaml and hazelcast.yaml present in root classpath
    contextRunner.run((context) -> {
        JetConfig jetConfig = context.getBean(JetInstance.class).getConfig();
        EdgeConfig defaultEdgeConfig = jetConfig.getDefaultEdgeConfig();
        Config hazelcastConfig = jetConfig.getHazelcastConfig();
        assertThat(defaultEdgeConfig.getQueueSize()).isEqualTo(2048);
        assertThat(hazelcastConfig.getClusterName()).isEqualTo("default-cluster");
    });
}
 

public static void main(String[] args) {
    JetInstance jet = JetBootstrap.getInstance();

    Configuration configuration = Configuration
            .create()
            .with("name", "mysql-demo-connector")
            .with("connector.class", "io.debezium.connector.mysql.MySqlConnector")
            /* begin connector properties */
            .with("database.hostname", "mysql")
            .with("database.port", "3306")
            .with("database.user", "debezium")
            .with("database.password", "dbz")
            .with("database.server.id", "184054")
            .with("database.server.name", "dbserver1")
            .with("database.whitelist", "inventory")
            .with("database.history.hazelcast.list.name", "test")
            .with("snapshot.mode", "schema_only")
            .build();

    Pipeline p = Pipeline.create();

    p.readFrom(DebeziumSources.cdc(configuration))
     .withoutTimestamps()
     .map(sourceRecord -> {
         String keyString = Values.convertToString(sourceRecord.keySchema(), sourceRecord.key());
         String valueString = Values.convertToString(sourceRecord.valueSchema(), sourceRecord.value());
         return Tuple2.tuple2(keyString, valueString);
     })
     .writeTo(Sinks.logger());

    jet.newJob(p).join();
}
 

private static JetInstance getJetInstance() {
    String bootstrap = System.getProperty("bootstrap");
    if (bootstrap != null && bootstrap.equals("true")) {
        return JetBootstrap.getInstance();
    }
    return Jet.newJetInstance();
}
 

public static void main(String[] args) {
    JetInstance jet = Jet.newJetInstance();
    Pipeline p = buildPipeline();

    System.out.println("Generating model...");
    try {
        jet.newJob(p).join();
        printTransitionsAndMarkovChain(jet);
    } finally {
        Jet.shutdownAll();
    }
}
 

/**
 * Prints state transitions from IMap, generates the markov chain and prints it
 */
private static void printTransitionsAndMarkovChain(JetInstance jet) {
    IMap<String, SortedMap<Double, String>> transitions = jet.getMap("stateTransitions");
    printTransitions(transitions);
    String chain = generateMarkovChain(1000, transitions);
    System.out.println(chain);
}
 

public static void main(String[] args) {
    System.out.println("DISCLAIMER: This is not investment advice");

    Pipeline pipeline = buildPipeline();
    // Start Jet
    JetInstance jet = Jet.newJetInstance();
    try {
        new CryptoSentimentGui(jet.getMap(MAP_NAME_JET_RESULTS));
        jet.newJob(pipeline).join();
    } finally {
        Jet.shutdownAll();
    }
}
 

public static void main(String[] args) {
    if (args.length != 2) {
        System.err.println("Missing command-line arguments: <model-file-path> <validation-input-data>");
        System.exit(1);
    }

    Path modelFile = Paths.get(args[0]).toAbsolutePath();
    Path inputFile = Paths.get(args[1]).toAbsolutePath();
    validateFileReadable(modelFile);
    validateFileReadable(inputFile);

    System.setProperty("hazelcast.logging.type", "log4j");

    JetInstance jet = Jet.newJetInstance();

    JobConfig jobConfig = new JobConfig();
    jobConfig.setName("h2o Breast Cancer Classification");
    jobConfig.attachFile(modelFile.toString(), "model");

    Job job = jet.newJob(buildPipeline(inputFile), jobConfig);

    try {
        job.join();
    } finally {
        jet.shutdown();
    }
}
 

public static void main(String[] args) {
    Pipeline p = buildPipeline();

    JetInstance jet = Jet.bootstrappedInstance();

    try {
        Job job = jet.newJob(p);
        job.join();
    } finally {
        jet.shutdown();
    }
}
 

public static void main (String[] args) {
    Pipeline p = buildPipeline();

    JetInstance jet = Jet.bootstrappedInstance();

    try {
        Job job = jet.newJob(p);
        job.join();
    } finally {
        jet.shutdown();
    }
}
 

public static void main (String[] args) {
    JetInstance jet = Jet.bootstrappedInstance();

    jet.getMap(LATEST_TRADES_PER_SYMBOL).addEntryListener(new TradeListener(), true);

    Pipeline p = buildPipeline();
    try {
        jet.newJob(p).join();
    } finally {
        jet.shutdown();
    }
}
 

public static void main(String[] args) {
    Pipeline p = buildPipeline();

    JetInstance jet = Jet.bootstrappedInstance();

    try {
        jet.newJob(p).join();
    } finally {
        jet.shutdown();
    }
}
 

public static void main(String[] args) {
    Pipeline p = buildPipeline();

    JetInstance jet = Jet.bootstrappedInstance();

    try {
        Job job = jet.newJob(p);
        job.join();
    } finally {
        jet.shutdown();
    }
}
 

public static void main(String[] args) {
    Pipeline p = buildPipeline();

    JetInstance jet = Jet.bootstrappedInstance();

    try {
        Job job = jet.newJob(p);
        job.join();
    } finally {
        jet.shutdown();
    }
}
 

public static void main (String[] args) {
    Pipeline p = buildPipeline();

    JetInstance jet = Jet.bootstrappedInstance();

    try {
        jet.newJob(p).join();
    } finally {
        jet.shutdown();
    }
}