org.apache.flink.runtime.execution.Environment.getUserClassLoader()方法的使用及代码示例

x33g5p2x  于2022-01-19 转载在 其他  
字(8.6k)|赞(0)|评价(0)|浏览(195)

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

Environment.getUserClassLoader介绍

[英]Returns the user code class loader
[中]返回用户代码类加载器

代码示例

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

  1. @Override
  2. public OperatorStateBackend createOperatorStateBackend(
  3.     Environment env,
  4.     String operatorIdentifier) throws Exception {
  5.   //the default for RocksDB; eventually there can be a operator state backend based on RocksDB, too.
  6.   final boolean asyncSnapshots = true;
  7.   return new DefaultOperatorStateBackend(
  8.       env.getUserClassLoader(),
  9.       env.getExecutionConfig(),
  10.       asyncSnapshots);
  11. }

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

  1. @VisibleForTesting
  2. public static <OUT> List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> createRecordWriters(
  3.     StreamConfig configuration,
  4.     Environment environment) {
  5.   List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> recordWriters = new ArrayList<>();
  6.   List<StreamEdge> outEdgesInOrder = configuration.getOutEdgesInOrder(environment.getUserClassLoader());
  7.   Map<Integer, StreamConfig> chainedConfigs = configuration.getTransitiveChainedTaskConfigsWithSelf(environment.getUserClassLoader());
  8.   for (int i = 0; i < outEdgesInOrder.size(); i++) {
  9.     StreamEdge edge = outEdgesInOrder.get(i);
  10.     recordWriters.add(
  11.       createRecordWriter(
  12.         edge,
  13.         i,
  14.         environment,
  15.         environment.getTaskInfo().getTaskName(),
  16.         chainedConfigs.get(edge.getSourceId()).getBufferTimeout()));
  17.   }
  18.   return recordWriters;
  19. }

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

  1. @Override
  2. public OperatorStateBackend createOperatorStateBackend(Environment env, String operatorIdentifier) throws Exception {
  3.   return new DefaultOperatorStateBackend(
  4.     env.getUserClassLoader(),
  5.     env.getExecutionConfig(),
  6.     true) {
  7.     @Nonnull
  8.     @Override
  9.     public RunnableFuture<SnapshotResult<OperatorStateHandle>> snapshot(
  10.       long checkpointId,
  11.       long timestamp,
  12.       @Nonnull CheckpointStreamFactory streamFactory,
  13.       @Nonnull CheckpointOptions checkpointOptions) throws Exception {
  14.       throw new Exception("Sync part snapshot exception.");
  15.     }
  16.   };
  17. }

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

  1. private RecordWriterOutput<OUT> createStreamOutput(
  2.     RecordWriter<SerializationDelegate<StreamRecord<OUT>>> recordWriter,
  3.     StreamEdge edge,
  4.     StreamConfig upStreamConfig,
  5.     Environment taskEnvironment) {
  6.   OutputTag sideOutputTag = edge.getOutputTag(); // OutputTag, return null if not sideOutput
  7.   TypeSerializer outSerializer = null;
  8.   if (edge.getOutputTag() != null) {
  9.     // side output
  10.     outSerializer = upStreamConfig.getTypeSerializerSideOut(
  11.         edge.getOutputTag(), taskEnvironment.getUserClassLoader());
  12.   } else {
  13.     // main output
  14.     outSerializer = upStreamConfig.getTypeSerializerOut(taskEnvironment.getUserClassLoader());
  15.   }
  16.   return new RecordWriterOutput<>(recordWriter, outSerializer, sideOutputTag, this);
  17. }

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

  1. @Override
  2. public OperatorStateBackend createOperatorStateBackend(Environment env, String operatorIdentifier) throws Exception {
  3.   return new DefaultOperatorStateBackend(
  4.     env.getUserClassLoader(),
  5.     env.getExecutionConfig(),
  6.     true) {
  7.     @Nonnull
  8.     @Override
  9.     public RunnableFuture<SnapshotResult<OperatorStateHandle>> snapshot(
  10.       long checkpointId,
  11.       long timestamp,
  12.       @Nonnull CheckpointStreamFactory streamFactory,
  13.       @Nonnull CheckpointOptions checkpointOptions) throws Exception {
  14.       return new FutureTask<>(() -> {
  15.         throw new Exception("Async part snapshot exception.");
  16.       });
  17.     }
  18.   };
  19. }

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

  1. public StreamingRuntimeContext(AbstractStreamOperator<?> operator,
  2.                 Environment env, Map<String, Accumulator<?, ?>> accumulators) {
  3.   super(env.getTaskInfo(),
  4.       env.getUserClassLoader(),
  5.       operator.getExecutionConfig(),
  6.       accumulators,
  7.       env.getDistributedCacheEntries(),
  8.       operator.getMetricGroup());
  9.   this.operator = operator;
  10.   this.taskEnvironment = env;
  11.   this.streamConfig = new StreamConfig(env.getTaskConfiguration());
  12.   this.operatorUniqueID = operator.getOperatorID().toString();
  13. }

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

  1. env.getUserClassLoader(),
  2. instanceBasePath,
  3. getDbOptions(),

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

  1. protected <K> InternalTimeServiceManager<K> internalTimeServiceManager(
  2.   AbstractKeyedStateBackend<K> keyedStatedBackend,
  3.   KeyContext keyContext, //the operator
  4.   Iterable<KeyGroupStatePartitionStreamProvider> rawKeyedStates) throws Exception {
  5.   if (keyedStatedBackend == null) {
  6.     return null;
  7.   }
  8.   final KeyGroupRange keyGroupRange = keyedStatedBackend.getKeyGroupRange();
  9.   final InternalTimeServiceManager<K> timeServiceManager = new InternalTimeServiceManager<>(
  10.     keyGroupRange,
  11.     keyContext,
  12.     keyedStatedBackend,
  13.     processingTimeService,
  14.     keyedStatedBackend.requiresLegacySynchronousTimerSnapshots());
  15.   // and then initialize the timer services
  16.   for (KeyGroupStatePartitionStreamProvider streamProvider : rawKeyedStates) {
  17.     int keyGroupIdx = streamProvider.getKeyGroupId();
  18.     Preconditions.checkArgument(keyGroupRange.contains(keyGroupIdx),
  19.       "Key Group " + keyGroupIdx + " does not belong to the local range.");
  20.     timeServiceManager.restoreStateForKeyGroup(
  21.       streamProvider.getStream(),
  22.       keyGroupIdx, environment.getUserClassLoader());
  23.   }
  24.   return timeServiceManager;
  25. }

代码示例来源：origin: org.apache.flink/flink-runtime_2.11

  1. /**
  2.  * Returns the user code class loader of this invokable.
  3.  *
  4.  * @return user code class loader of this invokable.
  5.  */
  6. public ClassLoader getUserCodeClassLoader() {
  7.   return getEnvironment().getUserClassLoader();
  8. }

代码示例来源：origin: org.apache.flink/flink-runtime

  1. /**
  2.  * Returns the user code class loader of this invokable.
  3.  *
  4.  * @return user code class loader of this invokable.
  5.  */
  6. public ClassLoader getUserCodeClassLoader() {
  7.   return getEnvironment().getUserClassLoader();
  8. }

代码示例来源：origin: com.alibaba.blink/flink-runtime

  1. /**
  2.  * Returns the user code class loader of this invokable.
  3.  *
  4.  * @return user code class loader of this invokable.
  5.  */
  6. public ClassLoader getUserCodeClassLoader() {
  7.   return getEnvironment().getUserClassLoader();
  8. }

代码示例来源：origin: org.apache.flink/flink-runtime_2.10

  1. /**
  2.  * Returns the user code class loader of this invokable.
  3.  *
  4.  * @return user code class loader of this invokable.
  5.  */
  6. public ClassLoader getUserCodeClassLoader() {
  7.   return getEnvironment().getUserClassLoader();
  8. }

代码示例来源：origin: org.apache.flink/flink-runtime_2.10

  1. @Override
  2. public OperatorStateBackend createOperatorStateBackend(
  3.   Environment env,
  4.   String operatorIdentifier) throws Exception {
  5.   return new DefaultOperatorStateBackend(
  6.     env.getUserClassLoader(),
  7.     env.getExecutionConfig(),
  8.     asynchronousSnapshots);
  9. }

代码示例来源：origin: org.apache.flink/flink-statebackend-rocksdb_2.10

  1. @Override
  2. public OperatorStateBackend createOperatorStateBackend(
  3.   Environment env,
  4.   String operatorIdentifier) throws Exception {
  5.   //the default for RocksDB; eventually there can be a operator state backend based on RocksDB, too.
  6.   final boolean asyncSnapshots = true;
  7.   return new DefaultOperatorStateBackend(
  8.     env.getUserClassLoader(),
  9.     env.getExecutionConfig(),
  10.     asyncSnapshots);
  11. }

代码示例来源：origin: org.apache.flink/flink-statebackend-rocksdb

  1. @Override
  2. public OperatorStateBackend createOperatorStateBackend(
  3.     Environment env,
  4.     String operatorIdentifier) throws Exception {
  5.   //the default for RocksDB; eventually there can be a operator state backend based on RocksDB, too.
  6.   final boolean asyncSnapshots = true;
  7.   return new DefaultOperatorStateBackend(
  8.       env.getUserClassLoader(),
  9.       env.getExecutionConfig(),
  10.       asyncSnapshots);
  11. }

代码示例来源：origin: org.apache.flink/flink-runtime_2.10

  1. @Override
  2. public OperatorStateBackend createOperatorStateBackend(
  3.   Environment env,
  4.   String operatorIdentifier) throws Exception {
  5.   return new DefaultOperatorStateBackend(
  6.     env.getUserClassLoader(),
  7.     env.getExecutionConfig(),
  8.     asynchronousSnapshots);
  9. }

代码示例来源：origin: org.apache.flink/flink-runtime_2.11

  1. @Override
  2. public OperatorStateBackend createOperatorStateBackend(
  3.   Environment env,
  4.   String operatorIdentifier) {
  5.   return new DefaultOperatorStateBackend(
  6.     env.getUserClassLoader(),
  7.     env.getExecutionConfig(),
  8.       isUsingAsynchronousSnapshots());
  9. }

代码示例来源：origin: com.alibaba.blink/flink-runtime

  1. @Override
  2. public OperatorStateBackend createOperatorStateBackend(
  3.   Environment env,
  4.   String operatorIdentifier) {
  5.   return new DefaultOperatorStateBackend(
  6.     env.getUserClassLoader(),
  7.     env.getExecutionConfig(),
  8.     isUsingAsynchronousSnapshots());
  9. }

代码示例来源：origin: org.apache.flink/flink-runtime_2.11

  1. @Override
  2. public OperatorStateBackend createOperatorStateBackend(
  3.     Environment env,
  4.     String operatorIdentifier) throws Exception {
  5.   return new DefaultOperatorStateBackend(
  6.       env.getUserClassLoader(),
  7.       env.getExecutionConfig(),
  8.       isUsingAsynchronousSnapshots());
  9. }

代码示例来源：origin: org.apache.flink/flink-streaming-java_2.10

  1. public StreamingRuntimeContext(AbstractStreamOperator<?> operator,
  2.                 Environment env, Map<String, Accumulator<?, ?>> accumulators) {
  3.   super(env.getTaskInfo(),
  4.       env.getUserClassLoader(),
  5.       operator.getExecutionConfig(),
  6.       accumulators,
  7.       env.getDistributedCacheEntries(),
  8.       operator.getMetricGroup());
  9.   this.operator = operator;
  10.   this.taskEnvironment = env;
  11.   this.streamConfig = new StreamConfig(env.getTaskConfiguration());
  12. }

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

Environment

相关文章

    查看更多

    热门标签

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

    最新文章

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