常用工具类

hadoop 常用操作工具类

ParquetUtil

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.format.converter.ParquetMetadataConverter;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.ParquetReader;
import org.apache.parquet.hadoop.example.GroupReadSupport;
import org.apache.parquet.hadoop.metadata.BlockMetaData;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
import org.apache.parquet.io.api.Binary;
import org.apache.parquet.schema.OriginalType;
import org.apache.parquet.schema.PrimitiveType;
import org.apache.parquet.schema.Type;
import org.apache.spark.sql.catalyst.util.DateTimeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author longzl
 */
public class ParquetUtil {

    private static Logger logger = LoggerFactory.getLogger(ParquetUtil.class);

    public final static String STRING_TYPE = "string";
    public final static String LONG_TYPE = "long";
    public final static String FLOAT_TYPE = "float";
    public final static String INTEGER_TYPE = "integer";
    public final static String BOOLEAN_TYPE = "boolean";
    public final static String DOUBLE_TYPE = "double";

    public final static String BYTE_TYPE = "byte";
    public final static String SHORT_TYPE = "short";
    public final static String DATE_TYPE = "date";
    public final static String DECIMAL_TYPE = "decimal";
    public final static String TIMESTAMP_TYPE = "timestamp";
    public final static String BINARY_TYPE = "binary";

    public static String convertParquetType2Normal(PrimitiveType.PrimitiveTypeName parquetType) {
        switch (parquetType) {
            case INT64:
                return LONG_TYPE;
            case BINARY:
                return STRING_TYPE;
            case FLOAT:
                return FLOAT_TYPE;
            case INT32:
                return INTEGER_TYPE;
            case INT96:
                return STRING_TYPE;
            case BOOLEAN:
                return BOOLEAN_TYPE;
            case DOUBLE:
                return DOUBLE_TYPE;
            default:
                throw new InvalidRequestParamException("parquetType not supported");
        }
    }

    public static String convertPrimitiveFiled(PrimitiveType fieldType) {
        PrimitiveType.PrimitiveTypeName typeName = fieldType.getPrimitiveTypeName();
        OriginalType originalType = fieldType.getOriginalType();

        switch (typeName) {
            case BOOLEAN: {
                return BOOLEAN_TYPE;
            }
            case FLOAT: {
                return FLOAT_TYPE;
            }

            case DOUBLE: {
                return DOUBLE_TYPE;
            }

            case INT32: {
                if (originalType == null) {
                    return INTEGER_TYPE;
                }

                switch (originalType) {
                    case INT_8:
                        return BYTE_TYPE;
                    case INT_16:
                        return SHORT_TYPE;
                    case INT_32:
                        return INTEGER_TYPE;
                    case DATE:
                        return DATE_TYPE;
                    default:
                        throw new InvalidRequestParamException("parquetType not supported");
                }
            }
            case INT64: {
                if (originalType == null) {
                    return LONG_TYPE;
                }
                switch (originalType) {
                    case INT_64:
                        return LONG_TYPE;
                    case DECIMAL:
                        return DECIMAL_TYPE;
                    default:
                        throw new InvalidRequestParamException("parquetType not supported");
                }
            }
            case INT96: {
                return TIMESTAMP_TYPE;
            }
            case BINARY: {
                if (originalType == null) {
                    return STRING_TYPE;
                }
                switch (originalType) {
                    case UTF8:
                    case ENUM:
                    case JSON:
                        return STRING_TYPE;
                    case BSON:
                        return BINARY_TYPE;
                    case DECIMAL:
                        return DECIMAL_TYPE;
                    default:
                        throw new InvalidRequestParamException("parquetType not supported");
                }
            }
            case FIXED_LEN_BYTE_ARRAY: {
                if (originalType == null) {
                    throw new InvalidRequestParamException("parquetType not supported");
                }
                switch (originalType) {
                    case DECIMAL:
                        return DECIMAL_TYPE;
                    default:
                        throw new InvalidRequestParamException("parquetType not supported");
                }
            }

            default: {
                throw new InvalidRequestParamException("parquetType not supported");
            }
        }
    }


    /**
     * 获取parquet文件夹的内容
     *
     * @param parentPath 文件夹
     * @param config
     * @param count      获取行数
     * @return
     * @throws IOException
     */
    public static List<List<Object>> headRecord(String parentPath, Configuration config, Integer count) throws
            IOException {
        FileSystem fs = null;
        Integer rowCount = 0;
        List<List<Object>> records = new ArrayList<>();
        try {
            fs = FileSystem.get(config);
            RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path(parentPath), true);

            while (listFiles.hasNext() && rowCount < count) {
                LocatedFileStatus fileStatus = listFiles.next();
                if (fileStatus.isFile()
                        && !fileStatus.getPath().toString().toUpperCase().endsWith("_SUCCESS")
                ) {

                    ParquetReader reader = null;
                    try {
                        Path path = fileStatus.getPath();
                        reader = ParquetReader.builder(new GroupReadSupport(), path).withConf(config).build();
                        for (Group line = (Group) reader.read();
                             line != null && rowCount++ < count;
                             line = (Group) reader.read()) {
                            records.add(getRecordOneLine(line));
                        }
                    } catch (Exception e) {
                        logger.warn(e.getMessage());
                        continue;
                    } finally {
                        if (reader != null) {
                            reader.close();
                        }
                    }
                }
            }

        } finally {
            if (fs != null) {
                fs.close();
            }
        }
        return records;
    }

    /**
     * @param parentPath 文件夹
     * @param config
     * @return
     * @throws IOException
     */
    public static List<Field> getSchema(String parentPath, Configuration config) throws IOException {
        FileSystem fs = null;

        try {
            fs = FileSystem.get(config);
            RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path(parentPath), true);

            while (listFiles.hasNext()) {
                LocatedFileStatus fileStatus = listFiles.next();
                if (fileStatus.isFile()
                        && !fileStatus.getPath().toString().toUpperCase().endsWith("_SUCCESS")
                ) {

                    try {
                        return getSchema(fileStatus.getPath(), config);
                    } catch (Exception e) {
                        logger.warn(e.getMessage());
                        continue;
                    }
                }
            }

        } finally {
            if (fs != null) {
                fs.close();
            }
        }

        return new ArrayList<>();
    }


    /**
     * 获取某一paruqet 文件路径,获取的parquet文件
     *
     * @param parentPath    文件夹
     * @param configuration
     * @return
     * @throws IOException
     */
    @Deprecated
    public static Path getOneParquetPath(String parentPath, Configuration configuration) throws IOException {
        Path oneParquet = null;
        Path maxCountPath = null;
        long maxCount = 0;
        FileSystem fs = null;
        Long rowCount = 0L;
        try {
            fs = FileSystem.get(configuration);
            RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path(parentPath), true);

            while (listFiles.hasNext()) {
                LocatedFileStatus fileStatus = listFiles.next();
                if (fileStatus.isFile()
                        && !fileStatus.getPath().toString().toUpperCase().endsWith("_SUCCESS")) {

                    rowCount = 0L;

                    ParquetMetadata parquetMetadata;
                    try {
                        parquetMetadata = ParquetFileReader.readFooter(configuration, fileStatus,
                                ParquetMetadataConverter.NO_FILTER);
                    } catch (RuntimeException e) {
                        logger.warn(e.getMessage());
                        continue;
                    }

                    // 统计parquet文件 行数,若文件行数大于30 则结束循环
                    for (BlockMetaData blockMetaData : parquetMetadata.getBlocks()) {
                        rowCount += blockMetaData.getRowCount();
                    }

                    if (rowCount > 30) {
                        oneParquet = fileStatus.getPath();
                        break;
                    } else {
                        if (maxCount < rowCount) {
                            maxCount = rowCount;
                            maxCountPath = fileStatus.getPath();
                        }
                    }
                }
            }

            if (rowCount <= 30) {
                oneParquet = maxCountPath;
            }
        } finally {
            if (fs != null) {
                fs.close();
            }
        }
        return oneParquet;
    }

    /**
     * @param parentPath    文件夹
     * @param configuration
     * @return
     * @throws IOException
     */
    public static List<Path> getParquetPathList(String parentPath, Configuration configuration) throws IOException {
        FileSystem fs = null;
        List<Path> pathList = new ArrayList<>();

        try {
            fs = FileSystem.get(configuration);
            new Path(parentPath);
            RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path(parentPath), true);
            while (listFiles.hasNext()) {
                LocatedFileStatus fileStatus = listFiles.next();
                if (fileStatus.isFile()
                        && !fileStatus.getPath().toString().toLowerCase().endsWith("_SUCCESS")) {
                    try {
                        ParquetFileReader.readFooter(configuration, fileStatus, ParquetMetadataConverter.NO_FILTER);
                        pathList.add(fileStatus.getPath());
                    } catch (RuntimeException e) {
                        if (!e.getMessage().contains("is not a Parquet file")) {
                            throw e;
                        }
                    }
                }
            }
        } finally {
            if (fs != null) {
                fs.close();
            }
        }
        return pathList;
    }

    /**
     * 统计 文件的大小、列数和行数
     *
     * @param tablePath
     * @return
     */
    public static Map<String, Long> statisticsParquet(String tablePath, Configuration conf) {
        logger.debug("统计表信息={} start", tablePath);
        Map<String, Long> map = new HashMap<>(3);
        Long rows = 0L;
        Long columns = 0L;
        Long fileSize = 0L;

        FileSystem fs = null;

        try {
            fs = FileSystem.get(conf);
            Path parquetFile;
            boolean isFirst = true;

            RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path(tablePath), true);
            while (listFiles.hasNext()) {
                LocatedFileStatus fileStatus = listFiles.next();
                if (fileStatus.isFile()
                        && !fileStatus.getPath().toString().toUpperCase().endsWith("_SUCCESS")) {

                    parquetFile = fileStatus.getPath();
                    fileSize += fileStatus.getLen();

                    ParquetFileReader parquetFileReader = null;
                    try {
                        parquetFileReader = new ParquetFileReader(conf, parquetFile, ParquetMetadataConverter.NO_FILTER);
                        rows += parquetFileReader.getRecordCount();

                        if (isFirst) {
                            columns = (long) parquetFileReader.getFileMetaData().getSchema().getFieldCount();
                            isFirst = false;
                        }
                    } catch (IOException e) {
                        logger.warn(e.getMessage());
                        continue;
                    } finally {
                        if (parquetFileReader != null) {
                            parquetFileReader.close();
                        }
                    }

                }
            }

        } catch (IOException e) {
            logger.warn(e.getMessage());
            map.put("rows", -1L);
            map.put("columns", -1L);
            map.put("fileSize", -1L);
        } finally {
            if (fs != null) {
                try {
                    fs.close();
                } catch (IOException e) {
                    logger.error(e.getMessage(), e);
                }
            }
        }

        map.put("rows", rows);
        map.put("columns", columns);
        map.put("fileSize", fileSize);
        logger.debug("统计表信息={} end", tablePath);
        return map;
    }

    /**
     * 获取文件 schema 信息
     *
     * @param parquetPath 单个文件
     * @param config
     * @return
     */
    public static List<Field> getSchema(Path parquetPath, Configuration config) throws IOException {
        List<Field> fields = new ArrayList<>();
        ParquetMetadata parquetMetadata = ParquetFileReader.readFooter(config, parquetPath, ParquetMetadataConverter
                .NO_FILTER);
        for (Type type : parquetMetadata.getFileMetaData().getSchema().getFields()) {
            if (!type.isPrimitive()) {
                throw new InternalServerException("不支持 非 Primitive 类型");
            }
            String javaType = convertPrimitiveFiled(type.asPrimitiveType());

            fields.add(new Field(type.getName(), javaType));
        }

        return fields;
    }


    /**
     * 获取 parquet 的记录
     *
     * @param parquetPath 单个文件
     * @param config
     * @param number
     * @return
     * @throws IOException
     */
    @Deprecated
    public static List<List<Object>> headRecord(Path parquetPath, Configuration config, Integer number) throws
            IOException {
        if (number > 1000) {
            throw new InternalServerException("读取行数超过1000，容易内存溢出");
        }
        List<List<Object>> records = new ArrayList<>();
        ParquetReader reader = null;
        try {
            reader = ParquetReader.builder(new GroupReadSupport(), parquetPath).withConf(config).build();
            for (Group line = (Group) reader.read();
                 line != null && number-- > 0L;
                 line = (Group) reader.read()) {
                records.add(getRecordOneLine(line));
            }
        } catch (IOException e) {
            if (reader != null) {
                reader.close();
            }
        } finally {
            if (reader != null) {
                reader.close();
            }
        }
        return records;
    }

    private static List<Object> getRecordOneLine(Group line) {
        List<Object> objects = new ArrayList<>();
        for (Type type : line.getType().getFields()) {
            objects.add(getValue(type, line));
        }
        return objects;
    }


    public static Object getValue(Type type, Group line) {
        Object o;
        if (!type.isPrimitive()) {
            throw new InternalServerException("不支持 非 Primitive 类型");
        }
        int fieldIndex = line.getType().getFieldIndex(type.getName());
        int index = 0;
        String javaType = convertPrimitiveFiled(type.asPrimitiveType());
        try {
            switch (javaType) {
                case STRING_TYPE: {
                    o = (line.getString(fieldIndex, index));
                    break;
                }
                case INTEGER_TYPE: {
                    o = (line.getInteger(fieldIndex, index));
                    break;
                }
                case LONG_TYPE: {
                    o = (line.getLong(fieldIndex, index));
                    break;
                }
                case DOUBLE_TYPE: {
                    o = (line.getDouble(fieldIndex, index));
                    break;
                }
                case FLOAT_TYPE: {
                    o = (line.getFloat(fieldIndex, index));
                    break;
                }
                case BOOLEAN_TYPE: {
                    o = (line.getBoolean(fieldIndex, index));
                    break;
                }
                case BINARY_TYPE: {
                    o = (line.getBinary(fieldIndex, index));
                    break;
                }
                case TIMESTAMP_TYPE: {
                    Binary int96 = line.getInt96(fieldIndex, index);
                    ByteBuffer order = int96.toByteBuffer().order(ByteOrder.LITTLE_ENDIAN);
                    Long timeOfDayNanos = order.getLong();
                    Integer julianDay = order.getInt();
                    Long timeStamp = DateTimeUtils.fromJulianDay(julianDay, timeOfDayNanos);
                    o = DateTimeUtils.timestampToString(timeStamp);
                    break;
                }
                case DATE_TYPE: {
                    int integer = line.getInteger(fieldIndex, index);
                    o = DateTimeUtils.dateToString(integer);
                    break;
                }
                //todo DECIMAL_TYPE
                //case DECIMAL_TYPE: {
                //    Binary binary = line.getBinary(fieldIndex, index);
                //}
                default:
                    throw new InternalServerException("暂不支持:" + javaType);
            }
        } catch (RuntimeException e) {
            if (e.getMessage().contains("element number " + index + " in group")) {
                o = null;
            } else {
                throw e;
            }
        }

        return o;
    }


}