有时候我们需要对很多的图片或文件进行处理，如数据集，如果一个一个图片处理就会非常麻烦，这时候我们可以采用批量操作。linux 自带的ImageMagick能够很方便地对图片进行批量处理，比如，批量图片格式转换，大小转换等等。在这里记录一下如何将成千上万的xxx.TIF文件（标签图像文件格式）批量转换成对应的xxx.png格式。因为自己用到了双三次下采样（上采用同），所以也记录了。

1. 先在终端进入到xxx.TIF所在文件夹，然后输入下列语句先在终端进入到xxx.TIF所在文件夹，然后输入下列语句
   
   这个语句表示把xxx.TIF文件（*.TIF）转换成png格式（-format png ）,然后把png文件缩小5倍（-size %20 **.png)，最后保存到相应的文件夹（-path /home/img/Desktop/RAISE_HR），其他的语句可以对应添加和移除。
   2.在这里介绍一下怎么把文件夹里成千上万的文件按文件名的尾数进行查找，并进行删除
   这里 find /home/img/Desktop/RAISE_HR |grep 0.png$表示找出相应文件夹中文件名以0结尾的png文件，| xargs rm -r对以上找出的文件执行删除。其他的语句也可以自行添加或删除。
   3.再放一段python代码，将文件夹里的所有文件批量进行双三次（bicubic）下采样操作
   
   目前就记录这些啦！最后，要说的是Linux的命令行真的是非常方便啊哈哈哈

Flink快速应用

通过一个单词统计的案例，快速上手应用Flink，进行流处理（Streaming）和批处理（Batch）

第 1 节 单词统计案例（批数据）

1.1 需求

统计一个文件中各个单词出现的次数，把统计结果输出到文件
步骤：
1、读取数据源
2、处理数据源
a、将读到的数据源文件中的每一行根据空格切分
b、将切分好的每个单词拼接1 c、根据单词聚合（将相同的单词放在一起）
d、累加相同的单词（单词后面的1进行累加）
3、保存处理结果

1.2 代码实现

引入依赖pom.xml

<dependencies>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-java -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.11.1</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-java -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.12</artifactId>
            <version>1.11.1</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.12</artifactId>
            <version>1.11.1</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-scala -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_2.12</artifactId>
            <version>1.11.1</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.12</artifactId>
            <version>1.11.1</version>
            <!--<scope>provided</scope>-->
        </dependency>
<dependencies>

Java程序

package com.lagou;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * 1、读取数据源
 * 2、处理数据源
 *  a、将读到的数据源文件中的每一行根据空格切分
 *  b、将切分好的每个单词拼接1
 *  c、根据单词聚合（将相同的单词放在一起）
 *  d、累加相同的单词（单词后面的1进行累加）
 * 3、保存处理结果
 */
public class WordCountJavaBatch {
    public static void main(String[] args) throws Exception {
        String inputPath="D:\\data\\input\\hello.txt";
        String outputPath="D:\\data\\output";

        //获取flink的运行环境
        ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();
        DataSet<String> text = executionEnvironment.readTextFile(inputPath);
        FlatMapOperator<String, Tuple2<String, Integer>> wordOndOnes = text.flatMap(new SplitClz());

        //(hello 1) (you 1) (hi 1) (him 1)
        UnsortedGrouping<Tuple2<String, Integer>> groupedWordAndOne = wordOndOnes.groupBy(0);
        //(hello 1) (hello 1)
        AggregateOperator<Tuple2<String, Integer>> out = groupedWordAndOne.sum(1);//1代表第1个元素

        out.writeAsCsv(outputPath, "\n", " ").setParallelism(1);//设置并行度
        executionEnvironment.execute();//人为调用执行方法

    }

    static class SplitClz implements FlatMapFunction<String,Tuple2<String,Integer>>{

        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
            String[] s1 = s.split(" ");
            for (String word:s1) {
                collector.collect(new Tuple2<String,Integer>(word,1));//发送到下游

            }

        }
    }
}

原文件
输出文件

scala程序

import org.apache.flink.api.scala._

object WordCountScalaBatch {
  def main(args: Array[String]): Unit = {
    val inputPath = "D:\\\\data\\\\input\\\\hello.txt"
    val outputPath = "D:\\data\\output"
    val environment: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
    val text: DataSet[String] = environment.readTextFile(inputPath)
    val out: AggregateDataSet[(String, Int)] = text.flatMap(_.split(" ")).map((_, 1)).groupBy(0).sum(1)
    out.writeAsCsv(outputPath,"\n", " ").setParallelism(1)
    environment.execute("scala batch process")
  }
}

第 2 节 单词统计案例（流数据）

2.1 需求

Socket模拟实时发送单词，使用Flink实时接收数据，对指定时间窗口内（如5s）的数据进行聚合统计，每隔1s汇总计算一次，并且把时间窗口内计算结果打印出来。

2.2 代码实现

scala程序

import org.apache.flink.streaming.api.scala._

object WordCountScalaStream {
  def main(args: Array[String]): Unit = {
    //处理流式数据
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment  //拿到运行环境
    val streamData: DataStream[String] = environment.socketTextStream("hadoop102", 7777)
    val out = streamData.flatMap(_.split(" ")).map((_, 1)).keyBy(0).sum(1)
    out.print()
    environment.execute()
  }

}

java程序

package com.lagou;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class WordCountJavaStream {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> dataStream = executionEnvironment.socketTextStream("linux121", 7777);
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = dataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                for (String word : s.split(" ")) {
                    collector.collect(new Tuple2<String, Integer>(word, 1));
                }
            }
        }).keyBy(0).sum(1);
        sum.print();
        executionEnvironment.execute();
    }
}

运行之前需要勾选

#控制台输入,7777表示端口
nc -lp 7777
#然后输入单词，查看统计