如何解决Impala用户定义的聚合函数C ++导致impalad崩溃
我的任务是使用C ++创建与Impala SQL一起使用的中值函数。我已经在笔记本电脑上安装了Cloudera的独立版本（5.16.2），所以我没有机会使生产服务器崩溃。我在尝试使其工作方面遇到困难。
我正在使用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 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,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};
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};
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复制到HDFS并使用impala-shell在Impala中创建函数：
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-shell中收到以下错误...
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,bool> std::_Rb_tree<double,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,bool> std::map<double,sMedianInfoValue,sMedianInfoValue> > >::insert<std::pair<double,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的小测试程序可以完美地工作，包括更新功能和中值的计算。
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。