title: Java Spark - UDF can have silent error?
body: <p>I had a bad UDF (note: <strong>NOT</strong> filter udf!) what basically randomly failed. When I executed this UDF with SQL on a temptable registered on a DataFrame it's seems executed correctly, I not even realized the issue till I used just some random input in the tests and I just checked if I get any result. When I developed a meaningful unit test I experienced that I have less output rows than the expected. The reason for this was the issue with my UDF, but it's strange that the error was silent. Or at least I can't see any Error in the log, I can't catch an exception it just executed, and when it failed on a row the result didn't contained the transformed row. (The issue was not data related).</p>
  
  <p>Is there any way to detect such an issue? Or this is the expected behavior of spark?</p>
  
  <p>(Spark 1.6.1)</p>
  
  <p>The UDF issue:
  Basically what happened, the UDF get a HashMap, and instead of copying the keySet I accidentally get the keySet by keySet(), and this is "Returns a Set view of the keys contained in this map." and I removed elements from this set, so I screwed up the HashMap, that caused the issue in the UDF. And it seems when this caused an error during processing the given field from the actual row, the whole row disappeared from the result DataFrame. This is more like a hint. </p>
  
  <p>UPDATE:
  Here's the (anonymized version) code where I experienced the behavior I explained.</p>
  
  <p>The UDF what caused the issue</p>
  
  <pre><code>public class AnonymUdf {
  public static Tuple2&lt;Set&lt;Long&gt;, Set&lt;Long&gt;&gt; processing(WrappedArray&lt;GenericRowWithSchema&gt; inputData, HashMap&lt;Long, List&lt;Long&gt;&gt; someInputMap, Long actionTime) {
      Set&lt;Long&gt; someSet = new HashSet&lt;&gt;();
      Set&lt;Long&gt; anotherSet = new HashSet&lt;&gt;();
      // This is the problematic line, keySet provide a reference; the input set should not change
      Set&lt;Long&gt; theInputKeySet = someInputMap.keySet();
      // This is the fix: to copy the key set
      // Set&lt;Long&gt; theInputKeySet = someInputMap.keySet().stream().collect(Collectors.toSet());
      if (someInputMap != null &amp;&amp; someInputMap.size() &gt; 0) {
          Iterator&lt;GenericRowWithSchema&gt; it = inputData.iterator();
          int idIdx = -1;
          int idTypeIdx = -1;
          int timeIdx = -1;
          while (it.hasNext()) {
              GenericRowWithSchema r = it.next();
              if (idIdx == -1) {
                  idIdx = r.fieldIndex("beacon_id");
              }
              if (idTypeIdx == -1) {
                  idTypeIdx = r.fieldIndex("beacon_type_id");
              }
              if (timeIdx == -1) {
                  timeIdx = r.fieldIndex("action_time");
              }
              Object o = r.get(idTypeIdx);
              Integer beaconIdType = o instanceof Long ? ((Long) o).intValue() : (Integer) o;
              if (someInputMap.containsKey(r.getLong(idIdx))) {
                  // if given beacon id is part of the beacon id universe (someInputMap)
                  if (2 == beaconIdType &amp;&amp; actionTime &lt;= r.getLong(timeIdx)) {
                      // if the beacon event is conversion (type 2) and happened recently
                      someSet.add(r.getLong(idIdx));
                  } else {
                      anotherSet.add(r.getLong(idIdx));
                  }
              }
          }
      }
      // And this screw up the someInputMap's keySet
      theInputSet.removeAll(someSet);
      theInputSet.removeAll(anotherSet);
  
      return new Tuple2&lt;&gt;(someSet, theInputSet);
  }
  }
  </code></pre>
  
  <p>Call the UDF:</p>
  
  <pre><code>...
  sqlContext.udf().register("myUdf", (WrappedArray&lt;GenericRowWithSchema&gt; ybe) -&gt;
          AnonymUdf.processing(input, map, time),
          DataTypes.createStructType(
          Arrays.asList(
          DataTypes.createStructField("result", DataTypes.createArrayType(DataTypes.LongType), false),
          DataTypes.createStructField("result2", DataTypes.createArrayType(DataTypes.LongType), false)
          ))
          );
  
  dataFrame1.registerTempTable("tempTable");
  DataFrame dataFrame2 = sqlContext.sql("SELECT id, ..." +
          "myUdf(field) AS result FROM tempTable");
  return dataFrame2;
  </code></pre>