标签: flatmap

你什么时候在RxJava中使用map vs flatMap？

比方说,我们想将包含JSON的文件映射到包含JSON的字符串中 -

使用map,我们必须以某种方式处理Exception.但是如何？:

Observable.from(jsonFile).map(new Func1<File, String>() {
    @Override public String call(File file) {
        try {
            return new Gson().toJson(new FileReader(file), Object.class);
        } catch (FileNotFoundException e) {
            // So Exception. What to do ?
        }
        return null; // Not good :(
    }
});

使用flatMap,它更加冗长,但我们可以将问题转发到Observables链中,如果我们选择其他地方甚至重试,则可以处理错误:

Observable.from(jsonFile).flatMap(new Func1<File, Observable<String>>() {
    @Override public Observable<String> call(final File file) {
        return Observable.create(new Observable.OnSubscribe<String>() {
            @Override public void call(Subscriber<? super String> subscriber) {
                try {
                    String json = new Gson().toJson(new FileReader(file), Object.class);

                    subscriber.onNext(json);
                    subscriber.onCompleted();
                } …

我一直在检查即将到来的Java update,即:Java 8 or JDK 8.是的,我很不耐烦,有很多新东西,但是,有一些我不明白的东西,一些简单的代码:

final Stream<Integer>stream = Stream.of(1,2,3,4,5,6,7,8,9,10);
stream.flatMap();

javadoc是

public <R> Stream<R> flatMap(Function<? super T,? extends Stream<? extends R>> mapper)

返回一个流,该流包含将此流的每个元素替换为通过将提供的映射函数应用于每个元素而生成的映射流的内容的结果.每个映射的流在其内容放入此流后关闭.(如果映射的流为空,则使用空流.)这是一个中间操作.

如果有人创建了一些简单的现实生活示例flatMap,如何在以前的java版本Java[6,7]中编写代码以及如何使用相同的例程编写代码,我将不胜感激Java 8.

我正在使用RxJava链接异步操作,我想将一些变量传递到下游:

Observable
   .from(modifications)
   .flatmap( (data1) -> { return op1(data1); })
   ...
   .flatmap( (data2) -> { 
       // How to access data1 here ?
       return op2(data2);
   })

这似乎是一种常见的模式,但我无法找到有关它的信息.

似乎这两个功能非常相似.它们具有相同的签名(接受rx.functions.Func1<? super T, ? extends Observable<? extends R>> func),并且它们的大理石图看起来完全相同.不能在这里粘贴图片,但这里是concatMap的一个,这里是flatMap的一个.在结果的描述中似乎存在一些细微的差别Observable,其中一个由concatMap包含由结合的Observable产生的flatMap项产生,而由包含由首先合并结果的Observable 产生的项产生,并且发出该合并的结果.

然而,这种微妙之处对我来说完全不清楚.任何人都可以更好地解释这种差异,并且理想地给出一些说明这种差异的例子.

我对flatMap有点困惑(添加到Swift 1.2)

假设我有一些可选类型的数组,例如

let possibles:[Int?] = [nil, 1, 2, 3, nil, nil, 4, 5]

在Swift 1.1中,我会做一个过滤器,然后是这样的地图:

let filtermap = possibles.filter({ return $0 != nil }).map({ return $0! })
// filtermap = [1, 2, 3, 4, 5]

我一直尝试使用flatMap这样做:

var flatmap1 = possibles.flatMap({
    return $0 == nil ? [] : [$0!]
})

和

var flatmap2:[Int] = possibles.flatMap({
    if let exercise = $0 { return [exercise] }
    return []
})

我更喜欢最后一种方法(因为我不必强制解包$0!...我害怕这些并且不惜一切代价避免它们)除了我需要指定数组类型.

是否有一个替代方法可以根据上下文计算出类型,但是没有强制解包？

我一直在浏览 Java 16 的新闻和源代码，并且遇到了名为mapMulti. 早期访问的JavaDoc说它类似于flatMap并且已经被批准用于完全相同的 Java 版本。

<R> Stream<R> mapMulti?(BiConsumer<? super T,?? super Consumer<R>> mapper)

• 如何使用此方法执行一对 0..n 映射？
• 新方法如何工作以及它与flatMap. 什么时候最好？
• mapper可以调用多少次？

我有一个对象列表,其中一些可以是集合.我想得到一个普通物体流.

List<Object> objects = List.of(1, 2, "SomeString", List.of(3, 4, 5, 6), 
    7, List.of("a", "b", "c"),
    List.of(8, List.of(9, List.of(10))));

我想获得一个包含元素的流.

1, 2, "SomeString", 3, 4, 5, 6, 7, "a", "b", "c", 8, 9, 10

我试过了

Function<Object, Stream<Object>> mbjectToStreamMapper = null; //define it. I have not figured it out yet!
objects.stream().flatMap(ObjectToStreamMapper).forEach(System.out::println);

我还检查了一个示例,该示例演示了如何使用递归函数来展平集合.但是,在此示例中.collect(Collectors.toList());用于保持中间结果.Collectors.toList()是一个终端操作,它将立即开始处理流.我想得到一个流,我可以稍后重复.

更新

我同意评论,有一个由不同性质的对象组成的流是一个可怕的想法.我只是为了简单起见写了这个问题.在现实生活中,我可以听到不同的事件,并从传入的流处理一些业务对象,其中一些可以发送对象流,其他 - 只是单个对象.

根据 Mozilla 开发者网站：

The flatMap() method first maps each element using a mapping function, then flattens the result into a new array. It is identical to a map followed by a flat of depth 1, but flatMap is often quite useful, as merging both into one method is slightly more efficient.

例子：

let arr = [1, 2, 4, 2, 3, 3, 4, 5, 5, 5, 8, 8, 9, 10];

const flatMap = arr.flatMap(x => x);
console.log(flatMap);

TypeError: arr.flatMap() is not …

我有一组文件.文件的路径保存在文件中,例如"all_files.txt".使用apache spark,我需要对所有文件进行操作并对结果进行处理.

我想要做的步骤是:

• 通过读取"all_files.txt"创建RDD
• 对于"all_files.txt"中的每一行(每行是某个文件的路径),将每个文件的内容读入一个RDD
• 然后做一个操作所有内容

这是我为此写的代码:

def return_contents_from_file (file_name):
    return spark.read.text(file_name).rdd.map(lambda  r: r[0])

def run_spark():
    file_name = 'path_to_file'

    spark = SparkSession \
        .builder \
        .appName("PythonWordCount") \
        .getOrCreate()

    counts = spark.read.text(file_name).rdd.map(lambda r: r[0]) \ # this line is supposed to return the paths to each file
        .flatMap(return_contents_from_file) \ # here i am expecting to club all the contents of all files
        .flatMap(do_operation_on_each_line_of_all_files) # here i am expecting do an operation on each line of all files

这是抛出错误:

第323行,在get_return_value中py4j.protocol.Py4JError:调用o25时发生错误.getnewargs.跟踪:py4j.Py4JException:方法 …

考虑以下代码：

public static void main(String[] args) {
    Stream.iterate(1, i -> i + 1)
            .flatMap(i -> Stream.of(i, i, i))
            .peek(System.out::println)
            .limit(4)
            .forEach(i -> {});
}

Java 8的输出：

1
1
1
2
2
2

在Java 11中：

1
1
1
2

这是Java 8中的错误或预期的行为，在11中进行了更改吗？

上面的代码只是演示不同行为的示例，但是更重要的含义是，以下代码在Java 11中显示1,2,3，但在Java 8中进入了无限循环：

    Stream.iterate(0, i -> i + 10)
            .flatMap(i -> Stream.iterate(i + 1, j -> j + 1))
            .limit(3)
            .forEach(System.out::println);