private def encodeFeature()

in spark/spark-tensorflow-connector/src/main/scala/org/tensorflow/spark/datasources/tfrecords/serde/DefaultTfRecordRowEncoder.scala [111:150]
40 lines of code
22 McCabe index (conditional complexity)

  private def encodeFeature(row: Row, structField: StructField, index: Int): Feature = {
    val feature = structField.dataType match {
      case IntegerType => Int64ListFeatureEncoder.encode(Seq(row.getInt(index).toLong))
      case BooleanType =>   Int64ListFeatureEncoder.encode(Seq(if (row.getBoolean(index)) 1.toLong else 0.toLong))
      case LongType => Int64ListFeatureEncoder.encode(Seq(row.getLong(index)))
      case FloatType => FloatListFeatureEncoder.encode(Seq(row.getFloat(index)))
      case DoubleType => FloatListFeatureEncoder.encode(Seq(row.getDouble(index).toFloat))
      case DecimalType() => FloatListFeatureEncoder.encode(Seq(row.getAs[Decimal](index).toFloat))
      case StringType => BytesListFeatureEncoder.encode(Seq(row.getString(index).getBytes))
      case BinaryType => BytesListFeatureEncoder.encode(Seq(row.getAs[Array[Byte]](index)))
      case ArrayType(BooleanType, _) =>
        Int64ListFeatureEncoder.encode(ArrayData.toArrayData(row.get(index)).toBooleanArray().map(if (_) 1.toLong else 0.toLong))
      case ArrayType(IntegerType, _)  =>
        Int64ListFeatureEncoder.encode(ArrayData.toArrayData(row.get(index)).toIntArray().map(_.toLong))
      case ArrayType(LongType, _) =>
        Int64ListFeatureEncoder.encode(ArrayData.toArrayData(row.get(index)).toLongArray())
      case ArrayType(FloatType, _) =>
        FloatListFeatureEncoder.encode(ArrayData.toArrayData(row.get(index)).toFloatArray())
      case ArrayType(DoubleType, _) =>
        FloatListFeatureEncoder.encode(ArrayData.toArrayData(row.get(index)).toDoubleArray().map(_.toFloat))
      case ArrayType(DecimalType(), _) =>
        val decimalArray = ArrayData.toArrayData(row.get(index)).toArray[Decimal](DataTypes.createDecimalType())
        FloatListFeatureEncoder.encode(decimalArray.map(_.toFloat))
      case ArrayType(StringType, _) =>
        BytesListFeatureEncoder.encode(ArrayData.toArrayData(row.get(index))
          .toArray[String](ObjectType(classOf[String])).map(_.getBytes))
      case ArrayType(BinaryType, _) =>
        BytesListFeatureEncoder.encode(ArrayData.toArrayData(row.get(index)).toArray[Array[Byte]](BinaryType))
      case VectorType => {
        val field = row.get(index)
        field match {
          case v: SparseVector => FloatListFeatureEncoder.encode(v.toDense.toArray.map(_.toFloat))
          case v: DenseVector => FloatListFeatureEncoder.encode(v.toArray.map(_.toFloat))
          case _ => throw new RuntimeException(s"Cannot convert $field to vector")
        }
      }
      case _ => throw new RuntimeException(s"Cannot convert field to unsupported data type ${structField.dataType}")
    }
    feature
  }