spark streaming 從kafka拉取資料，根絕訊息內容，需要將不容的訊息放到不同的資料夾下，大致內容為 從訊息中拆分出域名，不同域名分不到不同目錄，域名下按訊息中的時間分年月日目錄，底層目錄下自定義檔案個數，實現追加
由於sparkstreaming 10秒執行一次job，需要重寫 OutputFormat，來實現按內容分目錄，檔案追加

val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)

val line = lines.map(x => (x, 1)).repartition
 
(20)  

line.saveAsHadoopFiles("","",classOf[Text],classOf[NullWritable],classOf[MyMultipleTextOutputFormat[Text,NullWritable]])

MyMultipleTextOutputFormat 即為我們重寫的類

package com.linkingcloud.bigdata.common;

import com.linkingcloud.bigdata.common.interpret.LineInterpret;
import org.apache.hadoop.fs.FileSystem;
import
 
 org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat;
import org.apache.hadoop.util.Progressable;

import org.apache.hadoop.mapred.RecordWriter;

import java.io.IOException;
import java.util.Iterator;
import java.util.Random;
import
 
 java.util.TreeMap;

/**
 * Created by apple on 2017/2/15.
 */
public class MyMultipleTextOutputFormat<K, V> extends MultipleTextOutputFormat<K, V> {

    private TextOutputFormat<K, V> theTextOutputFormat = null;

    public RecordWriter getRecordWriter(final FileSystem fs, final JobConf job, final String name, final Progressable arg3) throws IOException {

        return new RecordWriter() {

            TreeMap<String, RecordWriter<K, V>> recordWriters = new TreeMap();

            public void write(Object key, Object value) throws IOException {

                //key中為訊息內容，value無意義
                String line = key.toString();

                //根據訊息內容，定義輸出路徑和輸出內容（同時清洗資料）
                String[] ss = LineInterpret.interpretLine(line, "/test/spark/kafka");

                if (ss != null && ss.length == 2) {

                    //name的最後兩位為jobid，同一個檔案只能同時允許一個job寫入，多個job寫一個檔案會報錯，將jobid作為檔名的一部分
                    //能解決此問題
                    String finalPath = ss[1] + "-" + name.substring(name.length() - 2);

                    RecordWriter rw = (RecordWriter) this.recordWriters.get(finalPath);

                    try {

                        if (rw == null) {

                            rw = getBaseRecordWriter(fs, job, finalPath, arg3);

                            this.recordWriters.put(finalPath, rw);
                        }

                        rw.write(ss[0], null);

                    } catch (Exception e) {
                        //一個週期內，job不能完成，下一個job啟動，會造成同時寫一個檔案的情況，變更檔名，新增字尾
                        this.rewrite(finalPath + "-", ss[0]);
                    }
                }
            }

            public void rewrite(String path, String line) {

                String finalPath = path + new Random().nextInt(10);

                RecordWriter rw = (RecordWriter) this.recordWriters.get(finalPath);

                try {

                    if (rw == null) {

                        rw = getBaseRecordWriter(fs, job, finalPath, arg3);

                        this.recordWriters.put(finalPath, rw);
                    }

                    rw.write(line, null);

                } catch (Exception e) {
                    //重試
                    this.rewrite(finalPath, line);
                }
            }

            public void close(Reporter reporter) throws IOException {
                Iterator keys = this.recordWriters.keySet().iterator();

                while (keys.hasNext()) {
                    RecordWriter rw = (RecordWriter) this.recordWriters.get(keys.next());
                    rw.close(reporter);
                }

                this.recordWriters.clear();
            }
        };
    }


    protected RecordWriter<K, V> getBaseRecordWriter(FileSystem fs, JobConf job, String path, Progressable arg3) throws IOException {
        if (this.theTextOutputFormat == null) {
            this.theTextOutputFormat = new MyTextOutputFormat();
        }

        return this.theTextOutputFormat.getRecordWriter(fs, job, path, arg3);
    }
}

MyTextOutputFormat中實現對存在的檔案進行append，而不是覆蓋

package com.linkingcloud.bigdata.common;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.util.ReflectionUtils;

import java.io.DataOutputStream;
import java.io.IOException;

/**
 * Created by apple on 2017/2/15.
 */

public class MyTextOutputFormat<K, V> extends TextOutputFormat<K, V> {

    public MyTextOutputFormat() {
    }

    @Override
    public RecordWriter<K, V> getRecordWriter(FileSystem ignored, JobConf job, String path, Progressable progress) throws IOException {

        String keyValueSeparator = job.get("mapreduce.output.textoutputformat.separator", "\t");

        CompressionCodec codec = ReflectionUtils.newInstance(GzipCodec.class, job);

        Path file = FileOutputFormat.getTaskOutputPath(job, path + codec.getDefaultExtension());

        FileSystem fs = file.getFileSystem(job);

        String file_path = path + codec.getDefaultExtension();

        Path newFile = new Path(FileOutputFormat.getOutputPath(job), file_path);

        FSDataOutputStream fileOut;

        if (fs.exists(newFile)) {
            fileOut = fs.append(newFile,4096,progress);
        } else {
            fileOut = fs.create(newFile, progress);
        }
        return new TextOutputFormat.LineRecordWriter(new DataOutputStream(codec.createOutputStream(fileOut)), keyValueSeparator);

    }
}

結果如下：

spark streaming 採用gzip壓縮，會導致derect memory洩露，暫時沒有找到好的解決方法，只能不使用壓縮，誰解決了此問題，可以留言，感謝！