DEV Community: shivamanipatil

Digital certificates And PKI

shivamanipatil — Sat, 11 Nov 2023 11:54:40 +0000

In this post, I will talk about public key cryptography, how encryption/decryption happens in asymmetric key cryptography, and how digital signatures and certificates are useful.

Public key cryptography

Also, known as asymmetric key cryptography. Asymmetric because there are 2 keys - public and private.
Public key is available to anyone in public. But the private key remains with the original owner.
Can be used to encrypt/decrypt data or to verify the authenticity of the data(digital signatures).
The thing to note is that public/private can be used interchangeably (they are similar in that sense) i.e. data encrypted by one can be decrypted by another. For the sake of convention, in this post, we will use the terms public key for encryption and private key for decryption.
RSA is an example of one such cryptosystem.

Encryption and decryption

In asymmetric key cryptography. The public key is available to the public.
The act of encrypting the data involves using some cryptographic algorithm with inputs as the public key and data being encrypted.
The end result is seemingly random-looking data. Obtaining original data from this random-looking data involves using the cryptographic algorithm with inputs as the private key and encrypted data.
Since public keys are available to the public, the sender can encrypt the data which can only be decrypted by the receiver.

Digital signatures

Digital signatures are generally used to verify the integrity of the data. Practically speaking it follows, that party A generates some data known as a digital signature for the actual data. And when sending the actual data this digital signature is also sent with it. Other parties now can actually verify that the received data is correct and is really sent by party A by using the digital signature.
The way this works using public key cryptography is :
1. The sender generates the digital signature(can be thought of as a process similar to encrypting) using its private key (using some crypto algo).
2. The receiver now has this digital signature of the data, the data and, the public key of the sender. We have already discussed that the public and private keys are interchangeable in that data encrypted by one can be decrypted using the other.
3. Due to this receiver can retrieve the data using the sender's public key and the digital signature. If the sent data by the sender and retrieved data are the same, then the verification is successful, suggesting that the data was indeed sent by the sender.
Practically, the same digital signature can be performed on the subset of the data (or even the hash of the data!) and we could still perform similar verification.

Digital certificate

This is an application of digital signatures. Used to establish the identity of the owner(of the certificate) and also its public key. Digital certificates will be associated with a pair of public and private keys. The server/owner will have the private key and the public key will be distributed along with the digital certificate.
A question arises, with just this information in the digital certificate, can some malicious party pose as another entity using a forged digital certificate? i.e. we have yet to establish/verify that the public key really belongs to the said owner.
To solve this problem we have issuers of the digital certificate, and the issuer itself has its public/private key pair. And their job is to verify the ownership and sign the digital certificate using their private key. This issuer's digital signature is also present in the digital certificate.
To understand this end-to-end on how an end user can verify the digital cert, consider this flow :
1. The user wants to verify the server A digital certificate. The certificate has server A public key, issuer A details, and issuer A digital signature.
2. Assume that the user trusts issuer A and also has its public key. Then it can verify using the issuer's digital signature that the server A digital certificate was indeed signed by the issuer thereby confirming if the digital certificate is valid.
3. It can be clearly seen here that a chain of trust is established, that is to verify the digital certificate of server A we have to verify/trust the issuer's digital signature.
The issuer here acts as the Certificate Authority(CA). And the job of a CA is to verify the identities and issue digital certificates for the entities. In our case, Issuer A was the CA for server A digital certificate. What this really means is that if you trust a CA then you automatically trust certificates signed/issued by this CA!
The process of signing a digital certificate by an issuer effectively means signing it using the issuer's private key or using the issuer's own digital certificate(both are the same technically, just different uses of language). FYI, the Digital certificate owner is also known as subject in technical terms.
X.509 is one of the standards for public key certificates.

Digital certificate types

Generally there are 3 types :
1. Root certificate :
  - These certificates are at the top of the chain of trust and therefore are self-signed (i.e. signed using their own private key).
  - These are also trust anchors because the trust for these is to be assumed and we cannot verify or derive the trust for these.
2. Intermediate certificate :
  - These are signed by the root certificate(same as root CA). e.g. issuer A in the above example.
  - To verify intermediate certificates, clients will use root CA cert public keys.
3. Leaf certificate :
  - Signed by the intermediate certificate (same as intermediate CA).
  - These are terminal nodes i.e. these cannot be used to issue new digital certificates. (The entity owning this is not a CA!)
  - The role of intermediate CA is to delegate/proxy the role of signing from root CA.
I am using CA and their digital certificates interchangeably here. Although the CA is the one actually owning its digital certificate (so technically not the same) the act of signing itself involves the CA digital certificate(or the private key).

Till now we have discussed how subjects can acquire digital certificates(for their public keys) from the CA. And how those digital certificates can be verified by other parties by establishing the chain of trust till root CA.

It is apparent that the applications (e.g. web browsers) need to trust some CA certs for which verification cannot be derived (e.g root CA certs). To facilitate the validation of these operating systems and applications have something called trust stores containing the information about CAs they can trust!

Hopefully, this gave you an overview of Public key cryptography, digital signatures, digital certificates and Certification authority. These concepts are widely used. One notable application is TLS, but that is a story for a different post.

What is Wi-Fi anyway?

shivamanipatil — Thu, 09 Nov 2023 17:20:58 +0000

You might ask what you mean by 'What is Wi-Fi?'😕 - I connect my device to the WiFi network and boom I am accessing the internet. Well, I held this abstraction for quite some time and the time had come to delve into a lower abstraction, because why not, there is no harm in knowing some background details. For the sake of not losing my mind, I wouldn't dive into super lower-level details nor I have looked into them 😃. But this should at least give you an understanding of how 'WiFi' operates.

Background

Let's clear (or loosely define) some background terms :

Network: Group of computers communicating and sharing information/resources with each other.
Wireless network: Network without physical connections(wires, cables etc.).
Physical Layer(TCP/IP model Layer 1): The actual medium between the nodes of a network. Maybe an easier thing to imagine would be how bits are sent as signals between the computer nodes are handled by this layer.
Data link layer(TCP/IP model Layer 2): This layer handles framing, addressing, reliable connectivity, error detection/correction, collision detection etc. It makes sense that when data is being transmitted through the physical layer, some upper layer needs to handle these problems for effective communication.
Frame: Unit of transport at the data link layer. Would consist of some headers and the actual data (payload) being sent. Headers would help in establishing the connection(source/destination MAC addresses) and some other control data.

General scenario

Consider what happens when you try to connect to a WiFi network for the first time through your phone.

You see the name of the wifi network listed in your phone's wifi settings. The technical term for this is actually called SSID (service set identifier).
Some networks would require you to put in a password(most home wifi networks) and others would not(some mall/coffee shop networks).
You click connect and you are able to access the internet (in most cases!).

So a simple question arises what exactly is called Wi-Fi here?

WiFi is a standard used to enable wireless networking. And it operates at the physical layer and the data link layer.
At the physical layer, it enables communication between the nodes using radio waves. At the data link layer, it mostly does the things mentioned in the background section.

How does my phone even know/discover that a wireless network exists and information about it?

Routers periodically broadcast information about their network. It contains information such as SSID (the name!), security, signal strength etc. Technically called WiFi beacon signal.
And since this broadcast is via radio waves anyone within the vicinity can receive them.
Your phone periodically checks for these signals and lists them. WiFi standards would have defined the structure of the signal in the first place. And since your phone supports the WiFi standard it can interpret it.

OK I was in a coffee shop and connected to a network without a password then what happened?

The client (mobile) node sends an association request to the router, the router acknowledges the request and welcomes the client onto the network. It would receive an IP address through DHCP servers.
Once the IP address is assigned the client is part of the network. The communication required by the client will be handled and routed by the router (e.g. internet access).
The concern here is that this was an open network so there is no encryption of any sort between the router and the client. Anyone can intercept the WiFi radio signal and read the unencrypted frame payload. You still might be secured if communicating over TLS (Transport layer security).
But communication happening lower than TLS or using different transport like UDP might still be affected (e.g. DNS queries over UDP).

Well, what happens when I connect to a protected WiFi network requiring some password?

In this case, the network would have some type of Security type standard configured. Some examples of it are WPA/WPA2/WPA3. Most probably, you would have seen these on your phone when connecting to a password-protected WiFi network. This information is also broadcast along with wifi beacon signals.
This password detail is sent along with the association request to the router. With that, a 4-way handshake occurs between the client and the router. The end result would be that the client and router would have the same secret key with them, which will be used to encrypt/decrypt (both with the same key!) the frame payload. Thus, securing our connection.
These WPA standards typically differ in how the handshakes occur and how the symmetric key is generated.

Got it, let's say I am connected to a protected WiFi network how does the overall flow look like now when the client wants to communicate with some node out there on the internet?

The client generates the layer 2 frames and the frame payload itself has layer 3 details (IP address details of the node on the internet). The frame payload is encrypted using the symmetric key generated via 4-way handshake.
The client after looking at the local routing table infers that to send the data onto the internet the network router is the default gateway. So the router's MAC address is filled in the layer 2 destination frame header. And the frame is sent over the physical layer (i.e. radio waves for wifi).
The router intercepts the frames decrypts the payload and inspects that the IP address belongs to some node over the internet. 4. It then repacks this IP packet into another frame with the destination being the ISP (of course there might be intermediaries).
And similar thing happens in reverse for the response received (for our client).

I know I have omitted many details, but the point was to get to a level where we have a general idea of how the stuff works. Hope, this is useful to people looking for more details on the working of the WiFi.

HyperLogLog : Memory vs Accuracy

shivamanipatil — Tue, 17 May 2022 15:51:53 +0000

Streaming and sketching algorithms
Count Distinct problem
Hyperloglog(HLL)
HLL improvements
Demo

Streaming and sketching algorithms

Algorithms for data streams that examine data a few times and have low memory requirements.
Storing individual elements is not possible in data streams as in most cases the size is enormous.
We try to get approximate or "good enough" estimates while storing the "sketch" of the data(i.e some representation of data) but not the data itself. The "sketch" size should be much less than the data size.
In short, we are doing a tradeoff between accuracy and memory.

Count Distinct problem

Finding the number of unique/distinct elements in a multiset(set with repeated elements).
e.g., unique visitors to a website and unique IP addresses through a router.
Simple solution to this could be to keep a map of every element and its count. The main drawback of this approach is the storage of the map, as in the worst case, it is O(n).
A sketching algorithm could prove helpful for this case as, in most cases, we would not care about 100% accuracy.

Hyperloglog(HLL)

Sketching algorithm designed to solve the count-distinct problem within acceptable error rate. Described in the paper "HyperLogLog: the analysis of near-optimal cardinality estimation algorithm", published by Flajolet, Fusy, Gandouet and Meunier in 2007. Improvements were proposed in "HyperLogLog in Practice: Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", published by Stefan Heulem, Marc Nunkesse and Alexander Hall.

Based on simple obervation : On average, a sequence of k consecutive zeros in binary form of a number will occur once in every 2^k distinct entries. E.g. if we have a large enough collection of fixed-width numbers (e.g. 32/64/128 bit) and while going through it and we find a number starting with k consecutive zeros in binary form, we can be almost sure that there are at least 2^k numbers in that collection.

Simple Estimator

To keep the input data uniform and evenly distributed we use a hash function. Now each entry is of fixed width(e.g 32/64/128 bits).
While going through the collection we simply keep the count of longest consecutive sequence of zeros.
Example of simple estimator (credits : engineering.fb.com) :

We will define a counter/estimator as data structure which holds longest consecutive sequence of zeros for a collection ( R ).
So, according to our theory this estimator will give estimate/cardinality as E = 2^R.
Problem with this is that estimates will be only in powers of 2 and there can be lot of variability as only a single value having largest sequence of zeros can affect the total estimate.

Multiple estimators and HLL

To solve the issue faced by simple estimator we will use multiple estimators(let's say of m number) and divide the input collection between them. So, now each estimator has its own ( R ). We get single estimate using harmonic mean of individual estimates of counters.

A estimator is basically just a register or a single variable holding longest sequence of consecutive zeros. Each estimator can be thought of as a bin. So we have m bin i.e m registers (or just array of size m really) to imitate m estimator.

To divide the input collection between the m estimators we can reserve some starting bits of hashed value for bin index and rest bits for counting longest sequence of consecutive zeros.

Illustration of this (credits : engineering.fb.com) :

This is the basis of the hyperloglog algorithm. The algorithm described in the paper with some error corrections is as follows idea behind it remains same though :

HLL parameters and storage analysis

From above illustration we can clearely see that only 12 KB storage is required for a very large ndistinct value and relatively small error rate.

HLL improvements

Improvements were proposed in "HyperLogLog in Practice: Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", published by Stefan Heulem, Marc Nunkesse and Alexander Hall :

Using 64 bit hash function
Estimating small cardinalities
Sparse representation

64 bit hash function

Hash function with L bits will distinguish at most 2^L values. Beyond 4 billion unique entries 32 bit hash function wouldn't be
useful. 64 bit hash function would be usable till 1.8 * 10^19 cardinalities (which should suffice for all cases).

Here you can see size increased by just m bits or 2 KB(if m = 2^14 with error=0.008125).

Estimating small cardinalities

Raw estimate of original hll algorithm gives large error for small cardinalities. To reduce the error original algorithm uses linear counting for cardinalities less than 5*m/2.

Improvement paper found out that the error for small cardinalities is due to a bias(i.e algorithm overestimates the cardinalities). They also could correct the bias by precomputing the bias values and substracting them from the raw estimate.

This figure from Improvement paper shows the bias for raw estimate. Also, it shows linear counting is still better than bias-correction for smaller cardinalities.

From this they concluded to perform linear counting till 11500 for p = 14 and bias correction after that.

Sparse representation

When n << m, then most of the bins(2^14 if p=14) are empty and memory is wasted. So instead we can store just pairs of (idx, R) values (idx - index of bin, R - longest consecutive zeros sequence observed by that bin).

And when size of these pairs increases than the normal(dense) mode of m bins then we can convert to the normal(dense) mode. Thing to note is that we can merge pairs with same idx and keep just the pair with largest R.

Demo

For the demo I will be using postgress_hyperloglog. This is a postgreSQL extension implementing hyperloglog estimator. You can follow the installation steps from the repo README.

Example 1:

Here, we are generating 1,10000 series and then doing hyperloglog_distinct aggregate on these.
hyperloglog_distinct aggregate just creates a hll estimator, adds column elements and returns the estimate.
The hll estimates 9998.401034851891 which is pretty close to 10000 cardinality.

Example 2:

You can serialize these estimators, save them and load them again to operate on later.

hyperloglog_accum - creates hll estimator, adds column elements and returns the serialized estimator.
Output in second picture is actually base64 encoded byte string of the estimator.

Example 3:

We can also get estimate from this accumulated result(example 2).

Internal hll data structure is binary compatible with base64 encoded byte string - so they are interoperable. This way we can perform hyperloglog_get_estimate operation on the hll estimator returned by hyperloglog_accum.

For other examples you can refer to sql references and try it out yourself.

References :

"HyperLogLog: the analysis of near-optimal cardinality estimation algorithm", published by Flajolet, Fusy, Gandouet and Meunier
"HyperLogLog in Practice: Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", published by Stefan Heulem, Marc Nunkesse and Alexander Hall
https://engineering.fb.com/2018/12/13/data-infrastructure/hyperloglog/
https://github.com/conversant/postgres_hyperloglog

Spark aggregation with native API's

shivamanipatil — Mon, 28 Feb 2022 13:26:23 +0000

Spark aggregation Overview
TypedImperativeAggregate[T] abstract class
Example

Spark aggregation Overview

User Defined Aggregate Functions can be used. But are restrictive and require workarounds even for basic requirements.
Aggregates are unevaluable expressions and cannot have eval and doGenCode method.
Basic requirement would be to use user defined java objects as internal spark aggregation buffer type.
And, passing extra arguments to aggregates e.g aggregate(col, 0.24)
Spark provides TypedImperativeAggregate[T] contract for such requirement (imperative as in expressed in terms of imperative initialize, update, and merge methods).

TypedImperativeAggregate[T] abstract class

case TestAggregation(child: Expression) 
  extends TypedImperativeAggregate[T]  {

  // Check input types
  override def checkInputDataTypes(): TypeCheckResult

  // Initialize T
  override def createAggregationBuffer(): T

  // Update T with row
  override def update(buffer: T, inputRow: InternalRow): T

  // Merge Intermediate buffers onto first buffer
  override def merge(buffer: T, other: T): T

  // Final value
  override def eval(buffer: T): Any 

  override def withNewMutableAggBufferOffset(newOffset: Int): TestAggregation 

  override def withNewInputAggBufferOffset(newOffset: Int): TestAggregation 

  override def children: Seq[Expression]

  override def nullable: Boolean

  // Datatype of output
  override def dataType: DataType

  override def prettyName: String

  override def serialize(obj: T): Array[Byte] 

  override def deserialize(bytes: Array[Byte]): T 
}

Example

case class Average holds count and sum of elements and also acts as internal aggregate buffer.
Aggregate takes in a numeric column and an extra argument n and return avg(column) * n.
In SparkSQL this will look like :

SELECT multiply_average(salary, 2) as average_salary FROM employees

Spark alchemy's NativeFunctionRegistration is used to register functions to spark.
Aggregate Code :

import com.swoop.alchemy.spark.expressions.NativeFunctionRegistration
import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.aggregate.TypedImperativeAggregate
import org.apache.spark.sql.types._

import java.io.{ByteArrayInputStream, ByteArrayOutputStream, ObjectInputStream, ObjectOutputStream}


case class Average(var sum: Long, var count: Long)

case class AvgTest(
                  child: Expression,
                  nExpression : Expression,
                  override val mutableAggBufferOffset: Int = 0,
                  override val inputAggBufferOffset: Int = 0)
  extends TypedImperativeAggregate[Average]  {

//  private lazy val n: Long = nExpression.eval().asInstanceOf[Long]
  def this(child: Expression) = this(child, Literal(1), 0, 0)
  def this(child: Expression, nExpression: Expression) = this(child, nExpression, 0, 0)

  override def checkInputDataTypes(): TypeCheckResult = {
    child.dataType match {
      case LongType => TypeCheckResult.TypeCheckSuccess
      case _ => TypeCheckResult.TypeCheckFailure(s"$prettyName only supports long input")
    }
  }

  override def createAggregationBuffer(): Average = {
    new Average(0, 0)
  }

  override def update(buffer: Average, inputRow: InternalRow): Average = {
    val value = child.eval(inputRow)
    buffer.sum += value.asInstanceOf[Long]
    buffer.count += 1
    buffer
  }

  override def merge(buffer: Average, other: Average): Average = {
    buffer.sum += other.sum
    buffer.count += other.count
    buffer
  }

  override def eval(buffer: Average): Any = {
    val n: Int = nExpression.eval().asInstanceOf[Int]
    ((buffer.sum*n)/(buffer.count))
  }

  override def withNewMutableAggBufferOffset(newOffset: Int): AvgTest =
    copy(mutableAggBufferOffset = newOffset)

  override def withNewInputAggBufferOffset(newOffset: Int): AvgTest =
    copy(inputAggBufferOffset = newOffset)

  override def children: Seq[Expression] = Seq(child, nExpression)

  override def nullable: Boolean = true

  // The result type is the same as the input type.
  override def dataType: DataType = child.dataType

  override def prettyName: String = "avg_test"

  override def serialize(obj: Average): Array[Byte] = {
    val stream: ByteArrayOutputStream = new ByteArrayOutputStream()
    val oos = new ObjectOutputStream(stream)
    oos.writeObject(obj)
    oos.close()
    stream.toByteArray
  }

  override def deserialize(bytes: Array[Byte]): Average = {
    val ois = new ObjectInputStream(new ByteArrayInputStream(bytes))
    val value = ois.readObject
    ois.close()
    value.asInstanceOf[Average]
  }
}

Driver code :

object TestAgg {
  object BegRegister extends NativeFunctionRegistration {
    val expressions: Map[String, (ExpressionInfo, FunctionBuilder)] = Map(
      expression[AvgTest]("multiply_average")
    )
  }
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("FirstDemo")
    val sc = new SparkContext(conf)
    val spark = SparkSession.builder().appName("Demo").config(conf).getOrCreate()


    BegRegister.registerFunctions(spark)
      val df = spark.read.json("src/test/resources/employees.json")
      df.createOrReplaceTempView("employees")
      df.show()
      /*
      +-------+------+
      |   name|salary|
      +-------+------+
      |Michael|  3000|
      |   Andy|  4500|
      | Justin|  3500|
      |  Berta|  4000|
      +-------+------+
       */
      val result = spark.sql("SELECT multiply_average(salary) as average_salary FROM employees")
      result.show()
    /*
      +--------------+
      |average_salary|
      +--------------+
      |          3750|
      +--------------+
     */
      val result1 = spark.sql("SELECT multiply_average(salary, 2) as average_salary FROM employees")
      result1.show()
      /*
      +--------------+
      |average_salary|
      +--------------+
      |          7500|
      +--------------+
     */
      val result2 = spark.sql("SELECT multiply_average(salary, 3) as average_salary FROM employees")
      result2.show()
      /*
      +--------------+
      |average_salary|
      +--------------+
      |         11250|
      +--------------+
      */
  }
}

Here, nExpression represents our n argument. Other lines are self-explanatory.

Spark Catalyst Optimizer and spark Expression basics

shivamanipatil — Mon, 28 Feb 2022 13:19:10 +0000

Overview
Trees
Rules
Expression
CodegenFallback
Example of spark native function using Unary expression

Spark Catalyst Overview

Core of Spark dataframe API and SQL queries.
Supports cost based and rule based optimization.
Built to be extensible :
- Adding new optimization techniques and features
- Extending the optimizier for custom use cases
At core it uses trees
On top of it various libraries are written for query processing, optimization and execution.

Trees

Trees in Catalyst consists of node objects.
Node - type and zero/more children
E.g If Literal(v: Int), Attribute(name: String), Add(l: TreeNode, r: TreeNode) are simple node types then x+(2+5) can be represented as Add(Attribute(x), Add(Literal(2), Literal(5))).

Rules

Function from a tree to another tree i.e modifying the tree.
Replacing a pattern matched subtree with transformation. e.g Add(Literal(2), Literal(5)) => Literal(7)
Transform method provided with catalyst tree e.g Recursively updates substructures to combine Literals

tree.transform {
  case Add(Literal(c1), Literal(c2)) => Literal(c1+c2)
}
So, x + (2+5) => x + 7

Expression

Executable node in catalyst tree. Take inputs and evaluates them.
Can generate java source from which can be used for evaluation (docodegen()).
Should be deterministic. Like pure functions.

scala> import org.apache.spark.sql.catalyst.expressions.Expression
scala> import org.apache.spark.sql.catalyst.expressions.{Literal, Add}

// Expression and eval
scala> val e : Add(Literal(3), Literal(4))
scala> e.eval()
res0: Any = 7

// Deterministic?
scala> e.deterministic
res1: Boolean = true

Type of Expression	Kind	Use
BinaryExpression	Abstract class	2 children
CodegenFallback	trait	Interpreted mode, no code generation
UnaryExpression	Abstract Class	1 child
LeafExpression	abstract class	No children
Unevaluable	trait	Cannot be evaluated to produce a value (neither in interpreted nor code-generated expression evaluations), e.g AggregateExpression

Expression contract :

package org.apache.spark.sql.catalyst.expressions

    // only required methods that have no implementation
    def dataType: DataType // Data type of the result of evaluating an expression

    * The default behavior is to call the eval method of the expression. Concrete expression
    * implementations should override this to do actual code generation.
    def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode

    * Interpreted (non-code-generated) expression evaluation
    * Slower than generated code('relative?')
    def eval(input: InternalRow = EmptyRow): Any
    def nullable: Boolean
}

CodegenFallback

Trait derived from Expression which allows expressions to not support java code generation and go full interpreted mode.
e.g

trait NoCodegenExp extends UnaryExpression with CodegenFallback {}

Example of spark native function using Unary expression

Here we will write a native function using Codegen and CodegenFallback.
Codegen example :

import org.apache.spark.sql.catalyst.expressions.{Expression, ImplicitCastInputTypes, UnaryExpression}
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.types.{DataType, DateType}
import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode}

// Returns beginning of month date for a date

case class BeginningOfMonth(startDate: Expression)

  extends UnaryExpression
    with ImplicitCastInputTypes {
  override def child: Expression = startDate

  override def inputTypes: Seq[DataType] = Seq(DateType)

  override def dataType: DataType = DateType

  // .eval calls nullSafeEval if input is non-null else it returns null
  override def nullSafeEval(date: Any): Any = {
    val level = DateTimeUtils.parseTruncLevel(UTF8String.fromString("MONTH"))
    DateTimeUtils.truncDate(date.asInstanceOf[Int], level)
  }

  override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
    val level = DateTimeUtils.parseTruncLevel(UTF8String.fromString("MONTH"))
    val dtu   = DateTimeUtils.getClass.getName.stripSuffix("$")
    defineCodeGen(ctx, ev, sd => s"$dtu.parseTruncLevel($sd, $level)")
  }

  override def prettyName: String = "beginning_of_month"
}

CodegenFallback example :

import org.apache.spark.sql.catalyst.expressions.{Expression, ImplicitCastInputTypes, UnaryExpression}
import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.types.{DataType, DateType}
import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode}

// Returns beginning of month date for a date
case class BeginningOfMonth(startDate: Expression)

  extends UnaryExpression
    with ImplicitCastInputTypes with CodegenFallback{
  override def child: Expression = startDate

  override def inputTypes: Seq[DataType] = Seq(DateType)

  override def dataType: DataType = DateType

  // .eval calls nullSafeEval if input is non-null else it returns null
  override def nullSafeEval(date: Any): Any = {
    val level = DateTimeUtils.parseTruncLevel(UTF8String.fromString("MONTH"))
    DateTimeUtils.truncDate(date.asInstanceOf[Int], level)
  }

//  override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
//    val level = DateTimeUtils.parseTruncLevel(UTF8String.fromString("MONTH"))
//    val dtu   = DateTimeUtils.getClass.getName.stripSuffix("$")
//    defineCodeGen(ctx, ev, sd => s"$dtu.parseTruncLevel($sd, $level)")
//  }

  override def prettyName: String = "beginning_of_month"
}

The can be registered and used in spark as :

  def main(args: Array[String]): Unit = {
    ....
    // Register the function
    object BegRegister extends NativeFunctionRegistration {
        val expressions: Map[String, (ExpressionInfo, FunctionBuilder)] = Map(
        expression[BeginningOfMonth]("beg_m")
        )
    }
    BegRegister.registerFunctions(spark)

    import spark.implicits._
    val df = Seq(
      (Date.valueOf("2020-01-15")),
      (Date.valueOf("2020-01-20")),
      (null)
    ).toDF("some_date")
    df.show()
    df.createTempView("dates")
    val dfVal = spark.sql("SELECT beg_m(some_date) from dates")
    dfVal.show()
  }

In codegenfallback example CodegenFallback trait is used and doGenCode() method is not required as eval(or nullSafeEval) used instead.

References :

Integrating Elasticsearch 7 into Django project

shivamanipatil — Thu, 30 Apr 2020 07:50:50 +0000

Recently, I had to migrate a Django project from elasticsearch 2.x.x using haystack to latest at that time Elasticsearch 7. It is a Django project maintained by the ns3 team. I wrote a medium article for same link. I hope this is useful for someone working with the same. Comments and suggestions are appreciated.

DEV Community: shivamanipatil

Digital certificates And PKI

Public key cryptography

Encryption and decryption

Digital signatures

Digital certificate

Digital certificate types

What is Wi-Fi anyway?

Background

General scenario

HyperLogLog : Memory vs Accuracy

Table of contents

Streaming and sketching algorithms

Count Distinct problem

Hyperloglog(HLL)

Simple Estimator

Multiple estimators and HLL

HLL parameters and storage analysis

HLL improvements

64 bit hash function

Estimating small cardinalities

Sparse representation

Demo

Example 1:

Example 2:

Example 3:

Spark aggregation with native API's

Table of contents

Spark aggregation Overview

TypedImperativeAggregate[T] abstract class

Example

Spark Catalyst Optimizer and spark Expression basics

Table of contents

Spark Catalyst Overview

Trees

Rules

Expression

CodegenFallback

Example of spark native function using Unary expression

Integrating Elasticsearch 7 into Django project