impala用户定义聚合函数c++崩溃impalad

pgvzfuti  于 2021-05-27  发布在  Hadoop
关注(0)|答案(0)|浏览(325)

我的任务是用c++创建一个中值函数，与impalasql一起使用。我在笔记本电脑上安装了一个独立版本（5.16.2）的cloudera，这样我就不会冒险破坏我们的生产服务器。我很难让它工作。
我用的是 std::map 与 double 作为关键，作为价值的结构：

struct sMedianInfoValue {
 int count; // count of the keys
 int iBegRowNbr; //if expanded out, the beginning row number
 int iEndRowNbr; //if expanded out, the ending row number
};

以下是完整的udaf：


# include <stdio.h>

# include <vector>

# include <algorithm>

# include <iostream>

# include <string.h>

# include <sstream>

# include <map>

# include <iterator>

# include <math.h>

# include "MIMedian.h"

using namespace std;

struct sMedianInfoValue {
 int count; // count of the keys
 int iBegRowNbr; //if expanded out, the beginning row number
 int iEndRowNbr; //if expanded out, the ending row number
};

// Structure used to pass a pointer to the vector.
struct DblMapStruct {
 std::map<double,sMedianInfoValue> mKeyValueMap;
};

//Initialize the iTotalRows variable.
int iTotalRows = 0;

//*----------------------------------------------------------------------------------------------*
//* Helper function ComputeMedian.                                                               *
//*----------------------------------------------------------------------------------------------*
//Update the row numbers based on the counts in the map
void UpdateRowNbrs(std::map<double,sMedianInfoValue> pmKeyValueMap) {

 //Create an iterator into the map starting at the first entry.
 std::map<double,sMedianInfoValue>::iterator it = pmKeyValueMap.begin();

 //The previous row's value.
 int iPrevRowValue = 1;
 int iCount;
 int iBegRowNbr;
 int iEndRowNbr;
 double dKey;

 //Loop through the map updating the beginning and ending row numbers.
 while (it != pmKeyValueMap.end()) {

  dKey = it->first;
  iCount = it->second.count;
  iTotalRows += iCount;
  iBegRowNbr = iPrevRowValue;
  iEndRowNbr = iBegRowNbr + iCount - 1;

  iPrevRowValue = iEndRowNbr + 1;

  //Update the map
  pmKeyValueMap[dKey] = {iCount,iBegRowNbr,iEndRowNbr};
  it++;

 }

}

//Compute the median from the map.
double ComputeMedian(std::map<double,sMedianInfoValue> pmKeyValueMap) {

 double dMedian = 0.0;
 string sEVENODD;
 int iFirstValue=0;
 int iSecondValue=0;
 double iFirstKey;
 double iSecondKey;

 //Based on iTotalRows, determine if it's even or odd.
 (iTotalRows & 1) ? sEVENODD="ODD" : sEVENODD="EVEN"; 

 //If EVEN, create one variable iTotalRows/2 and another iTotalRows/2 + 1;
 //If ODD, create one variable FLOOR(iTotalRows/2) and another CEIL(iTotalRows/2);
 if (sEVENODD=="EVEN") {
  iFirstValue = iTotalRows/2;
  iSecondValue = 1 + iFirstValue;
 } 
 else if (sEVENODD=="ODD") {
  iFirstValue = floor(iTotalRows/2);
  iSecondValue = 1 + iFirstValue;
 }

 //Iterate through the map finding the key value corresponding to the range of beginning and ending values.
 std::map<double,sMedianInfoValue>::iterator it = pmKeyValueMap.begin();
 while (it != pmKeyValueMap.end()) {

  if ( (it->second.iBegRowNbr <= iFirstValue) && (iFirstValue <= it->second.iEndRowNbr) ) {
   iFirstKey=it->first;
  } 

  if ( (it->second.iBegRowNbr <= iSecondValue) && (iSecondValue <= it->second.iEndRowNbr) ) {
   iSecondKey=it->first;
  } 

  it++;
 }

 //Compute the average for the two keys.
 dMedian = (iFirstKey + iSecondKey)/2;

 return dMedian;

}

//*----------------------------------------------------------------------------------------------*
//* UDAF Initialization function.                                                                *
//*----------------------------------------------------------------------------------------------*
void MIMedianInit(FunctionContext* context, StringVal* val) {

  val->ptr = context->Allocate(sizeof(DblMapStruct));

  if (val->is_null) {
    *val = StringVal::null();
    return;
  }

  val->is_null = false;
  val->len = sizeof(DblMapStruct);
  memset(val->ptr, 0, val->len);

}

//*----------------------------------------------------------------------------------------------*
//* UDAF Update function.                                                                        *
//*----------------------------------------------------------------------------------------------*
void MIMedianUpdate(FunctionContext* context, const DoubleVal& input, StringVal* val) {

  if (input.is_null || val->is_null) return;

  // Reconstitute the structure of the map.
  DblMapStruct* map = reinterpret_cast<DblMapStruct*>(val->ptr);

  // Insert the data into the map.
  int iCount=0;
  double dKey = input.val;

  //Update the map
  //Key not found in map.
  if (map->mKeyValueMap.find(dKey) == map->mKeyValueMap.end()) {

   //Key not found in map.
   iCount=1;
   sMedianInfoValue str_tmp = {iCount,0,0};
   map->mKeyValueMap.insert(std::make_pair(dKey,str_tmp));

  }
  else {

   //Key found in map...must update count.
   map->mKeyValueMap.at(dKey).count++;

  }

}

//*----------------------------------------------------------------------------------------------*
//* UDAF Merge function.                                                                         *
//*----------------------------------------------------------------------------------------------*
void MIMedianMerge(FunctionContext* context, const StringVal& src, StringVal* dst) {

 if (src.is_null || dst->is_null) return;

  // Reconstitute the structures of the maps.
  DblMapStruct* map_src = reinterpret_cast<DblMapStruct*>(src.ptr);
  DblMapStruct* map_dst = reinterpret_cast<DblMapStruct*>(dst->ptr);
  double dKey_src;
  int iCount_src;
  int iCount_dst;

  //Iterate source map and insert/update destination map.
  std::map<double,sMedianInfoValue>::iterator it = map_src->mKeyValueMap.begin();
  while (it != map_src->mKeyValueMap.end()) {

   dKey_src = it->first;
   iCount_src = it->second.count;
   iCount_dst = 0;

   //Update the map
   if (map_dst->mKeyValueMap.find(dKey_src) == map_dst->mKeyValueMap.end()) {

    //Key not found in map.
    iCount_dst=1;
    sMedianInfoValue str_tmp = {iCount_dst,0,0};
    map_dst->mKeyValueMap.insert(std::make_pair(dKey_src,str_tmp));

   }
   else {

    //Key found in map...get the existing count for this key.
    iCount_dst=map_dst->mKeyValueMap.at(dKey_src).count;
    map_dst->mKeyValueMap.at(dKey_src).count = iCount_src + iCount_dst;

   }

   it++;
  } 

}

//*----------------------------------------------------------------------------------------------*
//* UDAF Serialize function.                                                                     *
//*----------------------------------------------------------------------------------------------*
StringVal MIMedianSerialize(FunctionContext* context,const StringVal& val) {
 if (val.is_null) return StringVal::null();

 StringVal sResult = StringVal::CopyFrom(context,val.ptr,val.len);
 context->Free(val.ptr);

 return sResult;

}

//*----------------------------------------------------------------------------------------------*
//* UDAF Finalize function.                                                                      *
//*----------------------------------------------------------------------------------------------*
DoubleVal MIMedianFinalize(FunctionContext* context, const StringVal& val) {
  if (val.is_null) return DoubleVal::null();

  //Pointer to the map structure.
  DblMapStruct* map_src = reinterpret_cast<DblMapStruct*>(val.ptr);

  //Update the beginning and ending row counts in the map structure (they're all set to zero now).
  UpdateRowNbrs(map_src->mKeyValueMap);

  // Compute the median.
  DoubleVal dvMedian(ComputeMedian(map_src->mKeyValueMap));

  context->Free(val.ptr);

  return dvMedian;
}

这里是 MIMedian.h :


# ifndef IMPALA_MEDIAN_H

# define IMPALA_MEDIAN_H

# include <impala_udf/udf.h>

# include <assert.h>

# include <sstream>

using namespace impala_udf;
using namespace std;

void MIMedianInit(FunctionContext*, StringVal*);
void MIMedianUpdate(FunctionContext*, const DoubleVal&, StringVal*);
void MIMedianMerge(FunctionContext*, const StringVal&, StringVal*);
StringVal MIMedianSerialize(FunctionContext*,const StringVal&);
DoubleVal MIMedianFinalize(FunctionContext*, const StringVal&);

# endif

函数编译成功，没有错误或警告：

rm -f MIMedian.cc.o
rm -f libmimedian.so
/usr/bin/c++ -fPIC -g -ggdb -std=c++11 -o MIMedian.cc.o -c MIMedian.cc
/usr/bin/c++ -fPIC -g -ggdb -std=c++11 -shared -Wl,-soname,libmimedian.so -o libmimedian.so MIMedian.cc.o -lImpalaUdf

我复制文件 libmimedian.so 在impala中使用 impala-shell :

create aggregate function miimedian(double) returns double intermediate string location '/user/hive/warehouse/udf/libmimedian.so' init_fn='MIMedianInit' update_fn='MIMedianUpdate' merge_fn='MIMedianMerge' finalize_fn='MIMedianFinalize' serialize_fn='MIMedianSerialize';

当我使用此函数运行sql查询时。。。

select grp,miimedian(col1) as median from tab3 group by grp order by 1;

…我在 Impala 壳里发现了以下错误。。。

Socket error 104: Connection reset by peer
[Not connected] >

看着镜子 /var/lib/impala 文件夹，我看到一个 hs_err_pid####.log 文件，并在日志文件的顶部指示以下内容。。。


# Problematic frame:

# C  [libstdc++.so.6+0x748aa]

…再往下。。。

Stack: [0x00007fa3926a9000,0x00007fa392ea9000],  sp=0x00007fa392ea70b8,  free space=8184k
Native frames: (J=compiled Java code, j=interpreted, Vv=VM code, C=native code)
C  [libstdc++.so.6+0x748aa]

C  [libmimedian.5498.0.so+0xce4d]  std::_Rb_tree<double, std::pair<double const, sMedianInfoValue>, std::_Select1st<std::pair<double const, sMedianInfoValue> >, std::less<double>, std::allocator<std::pair<double const, sMedianInfoValue> > >::_M_get_insert_unique_pos(double const&)+0xfd

C  [libmimedian.5498.0.so+0xc27c]  std::pair<std::_Rb_tree_iterator<std::pair<double const, sMedianInfoValue> >, bool> std::_Rb_tree<double, std::pair<double const, sMedianInfoValue>, std::_Select1st<std::pair<double const, sMedianInfoValue> >, std::less<double>, std::allocator<std::pair<double const, sMedianInfoValue> > >::_M_insert_unique<std::pair<double, sMedianInfoValue> >(std::pair<double, sMedianInfoValue>&&)+0x44

C  [libmimedian.5498.0.so+0xbd70]  std::pair<std::_Rb_tree_iterator<std::pair<double const, sMedianInfoValue> >, bool> std::map<double, sMedianInfoValue, std::less<double>, std::allocator<std::pair<double const, sMedianInfoValue> > >::insert<std::pair<double, sMedianInfoValue>, void>(std::pair<double, sMedianInfoValue>&&)+0x2e

C  [libmimedian.5498.0.so+0xb382]  MIMedianUpdate(impala_udf::FunctionContext*, impala_udf::DoubleVal const&, impala_udf::StringVal*)+0xe1

我完全不知所措！我用std：：map编写的小测试程序运行良好，包括更新函数和中值的计算。

hadoopimpalaAggregatec++cloudera-cdh

暂无答案！

目前还没有任何答案，快来回答吧！

我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com