文档章节

Spark 获取文本内最大的前3个数字实例

别寒
 别寒
发布于 2017/07/24 17:19
字数 161
阅读 66
收藏 0
package cn.hhb.spark.core;

import org.apache.spark.Accumulator;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;

/**
 * Created by dell on 2017/7/13.
 */
public class Top3 {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setAppName("AccumulatorVariable")
                .setMaster("local")
                .set("spark.testing.memory", "2147480000");

        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<String> lines = sc.textFile("C://number.log");
        JavaPairRDD<Integer, String> pairs = lines.mapToPair(new PairFunction<String, Integer, String>() {
            @Override
            public Tuple2<Integer, String> call(String s) throws Exception {
                return new Tuple2<Integer, String>(Integer.valueOf(s), s);
            }
        });

        JavaPairRDD<Integer, String> sortPairs = pairs.sortByKey(false);
        JavaRDD<Integer> sortNumbers = sortPairs.map(new Function<Tuple2<Integer, String>, Integer>() {
            @Override
            public Integer call(Tuple2<Integer, String> v1) throws Exception {
                return v1._1;
            }
        });

        sortNumbers.foreach(new VoidFunction<Integer>() {
            @Override
            public void call(Integer v1) throws Exception {
                System.out.println(v1);
            }
        });

        List<Integer> sortedNumberList = sortNumbers.take(3);
        for (Integer number : sortedNumberList){
            System.out.println(number);
        }

        sc.close();
    }
}

© 著作权归作者所有

共有 人打赏支持
别寒
粉丝 30
博文 271
码字总数 137605
作品 0
永州
程序员
[Spark]Spark Streaming 指南一 Example

1. 概述 Spark streaming是Spark核心API的一个扩展,它对实时流式数据的处理具有可扩展性、高吞吐量、可容错性等特点。数据可以从诸如Kafka,Flume,Kinesis或TCP套接字等许多源中提取,并且...

sjf0115
2017/03/01
0
0
一文理清Apache Spark内存管理脉络

Spark 作为一个基于内存的分布式计算引擎,其内存管理模块在整个系统中扮演着非常重要的角色。理解 Spark 内存管理的基本原理,有助于更好地开发 Spark 应用程序和进行性能调优。本文旨在梳理...

技术小能手
05/31
0
0
[Spark]Spark Streaming 指南四 输入DStreams和Receivers

1. 输入DStream与Receiver 输入DStreams表示从源中获取输入数据流的DStreams。在指南一示例中,lines表示输入DStream,它代表从netcat服务器获取的数据流。每一个输入DStream(除 file strea...

sjf0115
2017/03/02
0
0
Spark笔记整理(二):RDD与spark核心概念名词

[TOC] Spark RDD 非常基本的说明,下面一张图就能够有基本的理解: Spark RDD基本说明 1、Spark的核心概念是RDD (resilient distributed dataset,弹性分布式数据集),指的是一个只读的,可分...

xpleaf
04/25
0
0
spark出现GC overhead limit exceeded和java heap space

spark执行任务时出现java.lang.OutOfMemoryError: GC overhead limit exceeded和java.lang.OutOfMemoryError: java heap space 最直接的解决方式就是在spark-env.sh中将下面两个参数调节的尽...

闵开慧
2014/10/14
0
1

没有更多内容

加载失败,请刷新页面

加载更多

你为什么在Redis里读到了本应过期的数据

一个事故的故事 晚上睡的正香突然被电话吵醒,对面是开发焦急的声音:我们的程序在访问redis的时候读到了本应过期的key导致整个业务逻辑出了问题,需要马上解决。 看到这里你可能会想:这是不...

IT--小哥
今天
2
0
祝大家节日快乐,阖家幸福! centos GnuTLS 漏洞

yum update -y gnutls 修复了GnuTLS 漏洞。更新到最新 gnutls.x86_64 0:2.12.23-22.el6 版本

yizhichao
昨天
5
0
Scrapy 1.5.0之选择器

构造选择器 Scrapy选择器是通过文本(Text)或 TextResponse 对象构造的 Selector 类的实例。 它根据输入类型自动选择最佳的解析规则(XML vs HTML): >>> from scrapy.selector import Sele...

Eappo_Geng
昨天
4
0
Windows下Git多账号配置,同一电脑多个ssh-key的管理

Windows下Git多账号配置,同一电脑多个ssh-key的管理   这一篇文章是对上一篇文章《Git-TortoiseGit完整配置流程》的拓展,所以需要对上一篇文章有所了解,当然直接往下看也可以,其中也有...

morpheusWB
昨天
5
0
中秋快乐!!!

HiBlock
昨天
3
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部