小编Kay*_*ayV的帖子

我的 kafka 集群有 3 个代理和几个主题，每个主题有 5 个分区。现在我想为分区设置复制因子。

我可以为 kafka 主题的分区设置的最大复制因子是多少？

我有一个应用程序，它使用嵌入式 janusgraph 和 cassandra 作为后端数据库。

以前，我使用 cassandrathrift 进行连接，并且工作正常。以下是旧配置：

storage.backend=cassandrathrift
storage.cassandra.keyspace=t_graph

但我在超时方面遇到了一些问题。因此，我将配置从 cassandrathrift 更改为 cql。这是新配置：

storage.backend=cql
storage.cql.keyspace=t_graph
storage.cql.read-consistency-level=ONE

而且，现在我收到以下错误：

> Caused by: org.springframework.beans.factory.BeanCreationException:
> Could not autowire field: private in.graph.services.GraphService
> in.graph.services.FollowService.graphService; nested exception is
> org.springframework.beans.factory.BeanCreationException: Error
> creating bean with name 'graphService': Invocation of init method
> failed; nested exception is java.lang.IllegalArgumentException: Could
> not instantiate implementation:
> org.janusgraph.diskstorage.cql.CQLStoreManager    at
> org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor$AutowiredFieldElement.inject(AutowiredAnnotationBeanPostProcessor.java:573)
>   at
> org.springframework.beans.factory.annotation.InjectionMetadata.inject(InjectionMetadata.java:88)
>   at
> org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor.postProcessPropertyValues(AutowiredAnnotationBeanPostProcessor.java:331)
>   ... 28 common frames omitted Caused by:
> …

我正在执行以下程序:

Stream.of("d2", "a2", "b1", "b3", "c")
.filter(s -> {
    System.out.println("filter: " + s);
    return true;
})
.forEach(s -> System.out.println("forEach: " + s));

我得到的输出是:

filter:  d2
forEach: d2
filter:  a2
forEach: a2
filter:  b1
forEach: b1
filter:  b3
forEach: b3
filter:  c
forEach: c

但是,我期待以下输出:

filter:  d2
filter:  a2
filter:  b1
filter:  b3
filter:  c 
forEach: d2
forEach: a2
forEach: b1
forEach: b3
forEach: c

意思是,首先filter应该完全执行forEach方法循环,然后方法循环应该已经开始.

有什么我做错了吗？

我试着搜索这个问题,但没有回答解释Java 8这样做的方法.任何身体都可以帮助我吗？

使用 scala 终端的 rdd.cache() 方法缓存的 RDD 被存储在内存中。

这意味着它将消耗 Spark 进程本身可用的部分内存。

话说回来，如果ram有限，而缓存的RDD越来越多，那么spark什么时候会自动清理rdd缓存占用的内存呢？

假设我们有一个 RDD，它被多次使用。因此，为了一次又一次地保存计算，我们使用 rdd.persist() 方法持久化了这个 RDD。

所以当我们持久化这个 RDD 时，计算 RDD 的节点将存储它们的分区。

那么现在假设，包含这个 RDD 持久分区的节点失败了，那么会发生什么？Spark 如何恢复丢失的数据？有没有复制机制？还是其他什么机制？

我正在学习 spring 5 webflux 和反应流。并且有新的HandlerFunctions和RouterFunctions来实现Http请求和响应。

并根据文件：

处理函数的注释对应部分是带有 @RequestMapping 的方法。

既然@RequestMapping非常容易处理、实现和理解，那么为什么需要更复杂和更困难的方法来通过这个 HandlerFunctions 和 RouterFunction 实用程序来处理 Http 请求和响应呢？

请建议。

根据 Mono 和 Flux 的定义，它们都代表一个异步数据序列，在订阅之前什么也不会发生。

并且有两大类出版商：热出版商和冷出版商。Mono 和 Flux 为每个订阅重新生成数据。如果未创建订阅，则永远不会生成数据。

另一方面，热门发布者不依赖于任何数量的订阅者。

这是我的冷流代码：

        System.out.println("*********Calling coldStream************");
        Flux<String> source = Flux.fromIterable(Arrays.asList("ram", "sam", "dam", "lam"))
                .doOnNext(System.out::println)
                .filter(s -> s.startsWith("l"))
                .map(String::toUpperCase);

        source.subscribe(d -> System.out.println("Subscriber 1: "+d));
        source.subscribe(d -> System.out.println("Subscriber 2: "+d));
        System.out.println("-------------------------------------");

这是输出：

*********Calling composeStream************
ram
sam
dam
lam
Subscriber 1: LAM
ram
sam
dam
lam
Subscriber 2: LAM
-------------------------------------

如何将上述冷流转换为热流？

我正在查看Spliterator的文档,根据它,Spliterator不是线程安全的:

尽管它们在并行算法中具有明显的实用性,但预计分裂器不是线程安全的.相反,使用分裂器的并行算法的实现应确保分裂器一次仅由一个线程使用.这通常很容易通过串行线程限制来实现,这通常是通过递归分解工作的典型并行算法的自然结果.

但是,在其进一步的文件中,其中陈述了对上述陈述的矛盾陈述:

源的结构干扰可以通过以下方式进行管理(大致降低合意性的顺序):

源管理并发修改.例如,java.util.concurrent.ConcurrentHashMap的键集是并发源.从源创建的Spliterator报告CONCURRENT的特征.

那么这是否意味着从线程安全的集合生成的Spliterator是线程安全的？这样对吗？

我是反应流新手，正在学习使用 concat/concatWith 方法组合两个发布者（具体来说是 Flux）。

我可以用 concat 方法做的所有事情，都可以使用 concatWith 方法来实现。这是我使用的示例案例：

        Mono<String> mono1 = Mono.just(" karan ");
        Mono<String> mono2 = Mono.just(" | verma ");
        Mono<String> mono3 = Mono.just(" | kv ");

        Flux<String> flux1 = Flux.just(" {1} ","{2} ","{3} ","{4} " );
        Flux<String> flux2 = Flux.just(" |A|"," |B| "," |C| ");

        // FLux emits item each 500ms
        Flux<String> intervalFlux1 = Flux.interval(Duration.ofMillis(1000))
                                        .zipWith(flux1, (i, string) -> string);

        // FLux emits item each 700ms       
        Flux<String> intervalFlux2 = Flux
                                .interval(Duration.ofMillis(1000))
                                .zipWith(flux2, (i, string) -> string);



        System.out.println("**************Flux …