Impala User-Defined Aggregate Function C++ Crashes impalad

  aggregate, c++, cloudera-cdh, hadoop, impala

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++

LEAVE A COMMENT