복붙노트

[HADOOP] 하이브에서 COLLECT_SET ()을 사용하여 중복을 유지 하시겠습니까?

HADOOP

하이브에서 COLLECT_SET ()을 사용하여 중복을 유지 하시겠습니까?

Hive에서 수집 된 집합에 복제본을 유지하거나 Hive가 다른 방법을 사용하여 제공하는 집계 컬렉션을 시뮬레이션하는 방법이 있습니까? 동일한 키가있는 열의 모든 항목을 배열에 집계하여 복제본으로 사용하려고합니다.

I.E. :

hash_id | num_of_cats
=====================
ad3jkfk            4
ad3jkfk            4
ad3jkfk            2
fkjh43f            1
fkjh43f            8
fkjh43f            8
rjkhd93            7
rjkhd93            4
rjkhd93            7

반환해야합니다 :

hash_agg | cats_aggregate
===========================
ad3jkfk   Array<int>(4,4,2)
fkjh43f   Array<int>(1,8,8)
rjkhd93   Array<int>(7,4,7)

해결법

  1. ==============================

    1.Hive 0.13.0 이후에 COLLECT_LIST (col)를 사용하십시오.

    Hive 0.13.0 이후에 COLLECT_LIST (col)를 사용하십시오.

    SELECT
    hash_id, COLLECT_LIST(num_of_cats) AS aggr_set
FROM
    tablename
WHERE
    blablabla
GROUP BY
    hash_id
;
  2. ==============================

    2.내장 된 것은 없지만 집계를 포함하여 사용자 정의 함수를 만드는 것은 나쁘지 않습니다. 유일한 거친 부분은 일반 유형으로 만들려고하지만 여기에는 수집 예제가 있습니다.

    내장 된 것은 없지만 집계를 포함하여 사용자 정의 함수를 만드는 것은 나쁘지 않습니다. 유일한 거친 부분은 일반 유형으로 만들려고하지만 여기에는 수집 예제가 있습니다.

    package com.example;

import java.util.ArrayList;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.StandardListObjectInspector;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;

public class CollectAll extends AbstractGenericUDAFResolver
{
    @Override
    public GenericUDAFEvaluator getEvaluator(TypeInfo[] tis)
            throws SemanticException
    {
        if (tis.length != 1)
        {
            throw new UDFArgumentTypeException(tis.length - 1, "Exactly one argument is expected.");
        }
        if (tis[0].getCategory() != ObjectInspector.Category.PRIMITIVE)
        {
            throw new UDFArgumentTypeException(0, "Only primitive type arguments are accepted but " + tis[0].getTypeName() + " was passed as parameter 1.");
        }
        return new CollectAllEvaluator();
    }

    public static class CollectAllEvaluator extends GenericUDAFEvaluator
    {
        private PrimitiveObjectInspector inputOI;
        private StandardListObjectInspector loi;
        private StandardListObjectInspector internalMergeOI;

        @Override
        public ObjectInspector init(Mode m, ObjectInspector[] parameters)
                throws HiveException
        {
            super.init(m, parameters);
            if (m == Mode.PARTIAL1)
            {
                inputOI = (PrimitiveObjectInspector) parameters[0];
                return ObjectInspectorFactory
                        .getStandardListObjectInspector((PrimitiveObjectInspector) ObjectInspectorUtils
                        .getStandardObjectInspector(inputOI));
            }
            else
            {
                if (!(parameters[0] instanceof StandardListObjectInspector))
                {
                    inputOI = (PrimitiveObjectInspector)  ObjectInspectorUtils
                            .getStandardObjectInspector(parameters[0]);
                    return (StandardListObjectInspector) ObjectInspectorFactory
                            .getStandardListObjectInspector(inputOI);
                }
                else
                {
                    internalMergeOI = (StandardListObjectInspector) parameters[0];
                    inputOI = (PrimitiveObjectInspector) internalMergeOI.getListElementObjectInspector();
                    loi = (StandardListObjectInspector) ObjectInspectorUtils.getStandardObjectInspector(internalMergeOI);
                    return loi;
                }
            }
        }

        static class ArrayAggregationBuffer implements AggregationBuffer
        {
            ArrayList<Object> container;
        }

        @Override
        public void reset(AggregationBuffer ab)
                throws HiveException
        {
            ((ArrayAggregationBuffer) ab).container = new ArrayList<Object>();
        }

        @Override
        public AggregationBuffer getNewAggregationBuffer()
                throws HiveException
        {
            ArrayAggregationBuffer ret = new ArrayAggregationBuffer();
            reset(ret);
            return ret;
        }

        @Override
        public void iterate(AggregationBuffer ab, Object[] parameters)
                throws HiveException
        {
            assert (parameters.length == 1);
            Object p = parameters[0];
            if (p != null)
            {
                ArrayAggregationBuffer agg = (ArrayAggregationBuffer) ab;
                agg.container.add(ObjectInspectorUtils.copyToStandardObject(p, this.inputOI));
            }
        }

        @Override
        public Object terminatePartial(AggregationBuffer ab)
                throws HiveException
        {
            ArrayAggregationBuffer agg = (ArrayAggregationBuffer) ab;
            ArrayList<Object> ret = new ArrayList<Object>(agg.container.size());
            ret.addAll(agg.container);
            return ret;
        }

        @Override
        public void merge(AggregationBuffer ab, Object o)
                throws HiveException
        {
            ArrayAggregationBuffer agg = (ArrayAggregationBuffer) ab;
            ArrayList<Object> partial = (ArrayList<Object>)internalMergeOI.getList(o);
            for(Object i : partial)
            {
                agg.container.add(ObjectInspectorUtils.copyToStandardObject(i, this.inputOI));
            }
        }

        @Override
        public Object terminate(AggregationBuffer ab)
                throws HiveException
        {
            ArrayAggregationBuffer agg = (ArrayAggregationBuffer) ab;
            ArrayList<Object> ret = new ArrayList<Object>(agg.container.size());
            ret.addAll(agg.container);
            return ret;
        }
    }
}

    그런 다음 하이브에서 단지 jar Whatever.jar를 추가하십시오. CREATE TEMPORARY FUNCTION collect_all AS 'com.example.CollectAll'; 예상대로 사용할 수 있어야합니다.

    hive> SELECT hash_id, collect_all(num_of_cats) FROM test GROUP BY hash_id;
OK
ad3jkfk [4,4,2]
fkjh43f [1,8,8]
rjkhd93 [7,4,7]

    요소의 순서는 정의되지 않은 것으로 간주해야한다는 점에 유의할 필요가 있습니다. 따라서 이것을 사용하여 정보를 n_gram으로 공급하려는 경우 필요에 따라 데이터를 정렬하기 위해 약간 확장해야 할 수도 있습니다.

  3. ==============================

    3.입력이 원시 타입이어야한다는 제한을 제거하기 위해 Jeff Mc의 코드를 수정했습니다 (아마도 collect_set에서 상속 됨). 이 버전은 구조체,지도 및 배열뿐만 아니라 프리미티브를 수집 할 수 있습니다.

    입력이 원시 타입이어야한다는 제한을 제거하기 위해 Jeff Mc의 코드를 수정했습니다 (아마도 collect_set에서 상속 됨). 이 버전은 구조체,지도 및 배열뿐만 아니라 프리미티브를 수집 할 수 있습니다.

    package com.example;

import java.util.ArrayList;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;
import org.apache.hadoop.hive.serde2.objectinspector.StandardListObjectInspector;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;

public class CollectAll extends AbstractGenericUDAFResolver
{
    @Override
    public GenericUDAFEvaluator getEvaluator(TypeInfo[] tis)
            throws SemanticException
    {
        if (tis.length != 1)
        {
            throw new UDFArgumentTypeException(tis.length - 1, "Exactly one argument is expected.");
        }
        return new CollectAllEvaluator();
    }

    public static class CollectAllEvaluator extends GenericUDAFEvaluator
    {
        private ObjectInspector inputOI;
        private StandardListObjectInspector loi;
        private StandardListObjectInspector internalMergeOI;

        @Override
        public ObjectInspector init(Mode m, ObjectInspector[] parameters)
                throws HiveException
        {
            super.init(m, parameters);
            if (m == Mode.PARTIAL1)
            {
                inputOI = parameters[0];
                return ObjectInspectorFactory
                        .getStandardListObjectInspector(ObjectInspectorUtils
                        .getStandardObjectInspector(inputOI));
            }
            else
            {
                if (!(parameters[0] instanceof StandardListObjectInspector))
                {
                    inputOI = ObjectInspectorUtils
                            .getStandardObjectInspector(parameters[0]);
                    return (StandardListObjectInspector) ObjectInspectorFactory
                            .getStandardListObjectInspector(inputOI);
                }
                else
                {
                    internalMergeOI = (StandardListObjectInspector) parameters[0];
                    inputOI = internalMergeOI.getListElementObjectInspector();
                    loi = (StandardListObjectInspector) ObjectInspectorUtils.getStandardObjectInspector(internalMergeOI);
                    return loi;
                }
            }
        }

        static class ArrayAggregationBuffer implements AggregationBuffer
        {
            ArrayList<Object> container;
        }

        @Override
        public void reset(AggregationBuffer ab)
                throws HiveException
        {
            ((ArrayAggregationBuffer) ab).container = new ArrayList<Object>();
        }

        @Override
        public AggregationBuffer getNewAggregationBuffer()
                throws HiveException
        {
            ArrayAggregationBuffer ret = new ArrayAggregationBuffer();
            reset(ret);
            return ret;
        }

        @Override
        public void iterate(AggregationBuffer ab, Object[] parameters)
                throws HiveException
        {
            assert (parameters.length == 1);
            Object p = parameters[0];
            if (p != null)
            {
                ArrayAggregationBuffer agg = (ArrayAggregationBuffer) ab;
                agg.container.add(ObjectInspectorUtils.copyToStandardObject(p, this.inputOI));
            }
        }

        @Override
        public Object terminatePartial(AggregationBuffer ab)
                throws HiveException
        {
            ArrayAggregationBuffer agg = (ArrayAggregationBuffer) ab;
            ArrayList<Object> ret = new ArrayList<Object>(agg.container.size());
            ret.addAll(agg.container);
            return ret;
        }

        @Override
        public void merge(AggregationBuffer ab, Object o)
                throws HiveException
        {
            ArrayAggregationBuffer agg = (ArrayAggregationBuffer) ab;
            ArrayList<Object> partial = (ArrayList<Object>)internalMergeOI.getList(o);
            for(Object i : partial)
            {
                agg.container.add(ObjectInspectorUtils.copyToStandardObject(i, this.inputOI));
            }
        }

        @Override
        public Object terminate(AggregationBuffer ab)
                throws HiveException
        {
            ArrayAggregationBuffer agg = (ArrayAggregationBuffer) ab;
            ArrayList<Object> ret = new ArrayList<Object>(agg.container.size());
            ret.addAll(agg.container);
            return ret;
        }
    }
}
  4. ==============================

    4.하이브 0.13부터 collect_list ()라는 UDAF가 내장되어 있습니다. 여길 봐.

    하이브 0.13부터 collect_list ()라는 UDAF가 내장되어 있습니다. 여길 봐.

  5. ==============================

    5.Brickhouse에서 UDAF (http://github.com/klout/brickhouse/blob/master/src/main/java/brickhouse/udf/collect/CollectUDAF.java)를 수집하십시오.

    Brickhouse에서 UDAF (http://github.com/klout/brickhouse/blob/master/src/main/java/brickhouse/udf/collect/CollectUDAF.java)를 수집하십시오.

    지도로 수집하는 기능도 지원합니다. Brickhouse에는 표준 Hive 배포본에없는 많은 유용한 UDF가 포함되어 있습니다.

  6. ==============================

    6.다음은이 작업을 수행하는 정확한 하이브 쿼리입니다 (하이브> 0.13에서만 작동).

    다음은이 작업을 수행하는 정확한 하이브 쿼리입니다 (하이브> 0.13에서만 작동).

    SELECT hash_id, collect_set (num_of_cats) FROM GROUP BY hash_id;

  7. ==============================

    7.Hive 0.13.0에는 중복 제거하지 않는 새로운 collect_list () 함수가 있습니다.

    Hive 0.13.0에는 중복 제거하지 않는 새로운 collect_list () 함수가 있습니다.

  8. ==============================

    8.구조체를 수집하는 해결 방법

    구조체를 수집하는 해결 방법

    테이블이 있다고 가정 해 보겠습니다.

    tableWithStruct(
id string,
obj struct <a:string,b:string>)

    이제 다른 테이블을 다음과 같이 만듭니다.

    CREATE EXTERNAL TABLE tablename (
id string,
temp array<string>
)
ROW FORMAT DELIMITED 
FIELDS TERMINATED BY '\t' COLLECTION ITEMS TERMINATED BY ',' MAP KEYS TERMINATED BY '|'

    검색어 삽입

    insert into table tablename select id,collect(concat_ws('|',cast(obj.a as string),cast(obj.b as string)) from tableWithStruct group by id;

    이제 tablename과 같은 위치에 다른 테이블을 만듭니다.

    CREATE EXTERNAL TABLE tablename_final (
id string,
array_list array<struct<a:string,b:string>>
)
ROW FORMAT DELIMITED 
FIELDS TERMINATED BY '\t' COLLECTION ITEMS TERMINATED BY ',' MAP KEYS TERMINATED BY '|'

    tablename_final에서 선택하면 원하는 출력을 얻을 수 있습니다.

  9. ==============================

    9.그냥 궁금해 -면 n 진술 -

    그냥 궁금해 -면 n 진술 -

    SELECT
    hash_id, COLLECT_LIST(num_of_cats) AS aggr_set
FROM
    tablename
WHERE
    blablabla
GROUP BY
    hash_id
;

    우리는 정렬하고 num_of_cats에 대한 요소를 제한하고 싶습니다 - 그것에 대해가는 방법? 대용량 데이터의 COZ 데이터의 PB를 처리합니다. 그런 경우에는 상위 10 개가 모두 필요하거나 제한하지 않을 수도 있습니다.

    10. from https://stackoverflow.com/questions/6445339/collect-set-in-hive-keep-duplicates by cc-by-sa and MIT license

'HADOOP' 카테고리의 다른 글

[HADOOP] Hadoop에서 노드를 올바르게 제거하려면 어떻게해야합니까?  (0) 2019.06.05
[HADOOP] SQL과 같은 하이브 삽입 쿼리  (0) 2019.06.05
[HADOOP] Hadoop 2.2 설치`. ' 그런 파일이나 디렉토리가 없다.  (0) 2019.06.05
[HADOOP] 아파치가 뾰족하게 달리지 않고 달릴 수 있습니까?  (0) 2019.06.05
[HADOOP] "hadoop fs"쉘 명령과 "hdfs dfs"쉘 명령의 차이점은 무엇입니까?  (0) 2019.06.05
,

티스토리툴바