I’ve been tasked to create a median function using C++ to be used with Impala SQL. I’ve installed a standalone version (5.16.2) of Cloudera on a laptop so I don’t take the chance of crashing our production servers. I’m having difficulty trying to get this to work.
I’m using a std::map with double as the key and a structure as the value:
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
};
Here is the complete 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;
}
Here’s 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
The function compiles successfully with no errors or warnings:
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
I copy the file libmimedian.so to HDFS and create the function in Impala using 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';
When I run a SQL query using this function…
select grp,miimedian(col1) as median from tab3 group by grp order by 1;
…I get the following error in impala-shell…
Socket error 104: Connection reset by peer
[Not connected] >
Looking in the /var/lib/impala folder, I see an hs_err_pid####.log file and it indicates the following at the top of the log file…
# Problematic frame:
# C [libstdc++.so.6+0x748aa]
…and further down…
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
I’m at a complete loss! My little test program using std::map works perfectly including the update function and the computation of the median.
Source: Windows Questions C++
