DEV Community: Ceyhun Kerti

Many to Many Relations in Ecto

Ceyhun Kerti — Thu, 23 Apr 2020 13:09:23 +0000

I believe this post will be helpful to the ones, searching for a many to many relationship implementation in phoenix/ecto.

I can't express enough my thoughts about elixir and surrounding ecosystem like phoenix and ecto, all i can say is it's a great pleasure to learn and be the part of the community even as a beginner. I have additional info about the community and resources at the end of the post, so let's get straight to the point and try to implement a many-to-many relation.

Suppose we have the classical blog application and our model is like;

Sure we have other columns like username, name etc.. in users or comment body in comments. I've skipped these additional attributes for the sake of simplicity.

Our migrations will look like this;

# users

defmodule MyApp.Repo.Migrations.AddUsersTable do
  use Ecto.Migration

  def change do
    create table(:users) do
      add :name, :string
      add :username, :string, null: false
      add :password_hash, :string, null: false
      add :email, :string, null: false

      timestamps()
    end

    create unique_index(:users, [:username])
    create unique_index(:users, [:email])

  end
end

# posts

defmodule MyApp.Repo.Migrations.CreatePosts do
  use Ecto.Migration

  def change do
    create table(:posts) do
      add :title, :string
      add :owner_id, references(:users, on_delete: :nothing), null: true

      timestamps()
    end
  end
end

# comments

defmodule MyApp.Repo.Migrations.CreateComments do
  use Ecto.Migration

  def change do
    create table(:comments) do
      add :body, :string
      add :post_id, references(:posts, on_delete: :delete_all)
      add :owner_id, references(:users, on_delete: :nothing), null: true

      timestamps()
    end

  end
end

# comment_likes

defmodule MyApp.Repo.Migrations.AddCommnetLikeTable do
  use Ecto.Migration

  def change do
    create table(:comment_likes, primary_key: false) do
      add(:comment_id, references(:comments, on_delete: :delete_all), primary_key: true)
      add(:user_id, references(:users, on_delete: :delete_all), primary_key: true)

      timestamps()
    end

    create(
      unique_index(:comment_likes, [:comment_id, :user_id], name: :comment_id_user_id_unique_index)
    )
  end
end

Now we add the schema definitions;

# User.ex

 schema "users" do
    field :name, :string
    field :password, :string, virtual: true
    field :password_hash, :string
    field :username, :string
    field :email, :string
    many_to_many :liked_comments, Comment, join_through: CommentLike, on_replace: :delete

    timestamps()
  end

# User.ex

 schema "users" do
    field :name, :string
    field :password, :string, virtual: true
    field :password_hash, :string
    field :username, :string
    field :email, :string
    many_to_many :liked_comments, Comment, join_through: CommentLike, on_replace: :delete

    timestamps()
  end

# Comment.ex

  schema "comments" do
    field :body, :string
    belongs_to :post, Post, references: :id
    belongs_to :owner, User, references: :id
    many_to_many :liked_by, User, join_through: CommentLike, on_replace: :delete

    timestamps()
  end

# Post.ex

  schema "posts" do
    field :title, :string
    belongs_to :owner, User, references: :id
    has_many :comments, Comment

    timestamps()
  end

finally we must also have a schema for the relation itself, i have skipped this part while i was first implementing it, just because i thought it was auto done, and no need for it but this was a mistake and I've learned it is actually necessary. So lets create it like;

# comment_like.ex

  @primary_key false
  schema "comment_likes" do
    belongs_to :user, User
    belongs_to :comment, Comment

    timestamps()
  end

OK the model is set, and the rest is easy; I'll give a sample implementation for liking a comment and getting comment list along with the attribute, is_liked which is if the current user is liked the comment or not.

Liking a comment;

# comment_controller.ex

def like(conn, %{"comment_id" => comment_id}) do
    case Context.like_comment(conn.assigns.current_user, comment_id) do
      {:ok, _} -> json conn, :ok
      {:error, _} -> json conn, :error
    end
  end

# context.ex


  def like_comment(current_user, comment_id) do
    current_user
    |> Repo.preload(:liked_comments)
    |> Ecto.Changeset.change
    |> Ecto.Changeset.put_assoc(:liked_comments, [get_comment!(comment_id)])
    |> Repo.update()
  end

Getting the list of comments for the post;

# context.ex

 def get_commnets(current_user_id, post_id) do
    query = from p in Post,
      join: c in assoc(p, :comments),
      left_join: cl in CommentLike,
      on: c.id == cl.comment_id,
      left_join: u in User,
      on: cl.user_id == u.id and u.id == ^current_user_id,
      where: p.id == ^post_id,
      select: %{comment: c, is_liked: not is_nil(u.id)}

    Repo.all(query)
  end

That's it.

I strongly suggest iex(elixir interactive shell) for trying, learning or debugging your ecto models. Just issue iex -S mix in your project folder and you are ready to go, you can interact with your application in console.

Lastly, elixir has a welcoming community. I've learned a lot from the posts and discord channel.

Here are a few resources for learning elixir and phoenix.

Thanks for reading i hope it'll help.

Streaming Analytics With Confluent Platform and Debezium

Ceyhun Kerti — Mon, 21 Oct 2019 16:23:30 +0000

This article is for those who wants to implement a real time analytics platform, but have the data stored initially in database. There are many documents on web describing, building a streaming platform, but most of them assume that the data is being ingested to Kafka in the first place. Unfortunately that's not the case for some of us.
Our example architectural scenario is like this;

Our data is initially stored in Oracle database, in other words application layer persists the user transactions directly to database instead of sending them to a queue like Kafka. So how do we convert the static data to a stream? Lucky for us there are some methods that utilizes the CDC(change data capture) logic. Since in our example scenario the database is Oracle we will only talk about the options for Oracle. Hopefully there are even more options for other databases for capturing real time data.

Currently there are two options for capturing data from Oracle database and the third one is on the way that we will talk about briefly later in this article.

GoldenGate

Oracle had acquired GoldenGate back in 2009 and positioned the software in their fusion middleware tier. Basically GoldenGate is a cdc tool that reads the oracle redo logs and replicates the data. Since it is reading from the redo logs it has no extra load on the database. Before GG some of the developers were performing trigger operations on the source systems and replicating the data over dblink to another oracle database, as expected this method had a serious drawbacks including performance, security and stability. A couple of years ago Oracle has released Oracle GoldenGate Big Data Adapter which you can configure a Kafka adapter and integrate GG with Kafka. That being said this adapter is tightly coupled with their BigData Appliance platform.

XStream API

XStream API is released as a new feature with Oracle database 11g Release 2 (11.2). Since then this feature is included in every other successor versions. XStream API basically consists of oracle database components, such as inbound and outbound servers, and an application programming interface API. This interface enable client applications to receive data changes from an Oracle database and send data changes to an Oracle database. Received changes can be shared with other other systems, including non Oracle products, file systems or third party software applications of any kind.

So which one should we use? The answer depends on your current architecture. If you are using Oracle BDA you may consider using GG Big Data adapter. Otherwise you may check out XStream API and learn how it works. One important thing should be noted here is; sadly both of the approaches require Oracle GoldenGate licence. Yes you read it correct, even if you use XStream you should have a GG licence.

Debezium

Alright lets get back to our sample scenario and ask ourselves; i don't have BDA and don't want to install GG, so which one should i use? Actually we do not use either of them directly. To come to our rescue, there is an open source RedHat Debezium platform for change data capture from various databases including Oracle. Debezium has a plugin for Oracle database which uses XStream API and simplifies many things for us. IMHO it is much more simpler then setting up GG and integrating with Kafka. But remember that, since it uses XStream API, you should have GG licence in any case. As we said before there is an upcoming third option ,aside from GG and XStream Api, inside Debezium which utilizes Oracle LogMiner that does not require a Oracle licence. This feature is being implemented as of writing this article, you may check github repo and gitter channel for more information. Especially i recommend gitter channel for any kind of issues and questions for Debezium. You can contact directly to product owners and developers, who are very friendly and helpful.

Integration

So let's get start to integrate Oracle, Debezium and Confluent. First of all we need to prepare our database so that we can receive change records.

Enable GoldenGate replication and archive log mode, you may need to change below paths according to your installation.

sqlplus sys as sysdba
alter system set db_recovery_file_dest_size = 5G;
mkdir -p /u01/oracle/oradata/recovery_area
alter system set db_recovery_file_dest = '/u01/oracle/oradata/recovery_area' scope=spfile;
alter system set ENABLE_GOLDENGATE_REPLICATION=true;
shutdown immediate
startup mount
alter database archivelog;
alter database open;
--Should show "Database log mode: Archive Mode"
archive log list

Create xstrm and xstrmadmin users for managing connections to oracle.

CREATE TABLESPACE xstream_adm_tbs DATAFILE '/u01/oracle/oradata/xstream_adm_tbs.dbf' SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

CREATE USER xstrmadmin IDENTIFIED BY xsa
DEFAULT TABLESPACE xstream_adm_tbs
QUOTA UNLIMITED ON xstream_adm_tbs;

GRANT CREATE SESSION TO xstrmadmin;

CREATE TABLESPACE xstream_tbs DATAFILE '/u01/oracle/oradata/xstream_tbs.dbf' SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

CREATE USER xstrm IDENTIFIED BY xs DEFAULT TABLESPACE xstream_tbs QUOTA UNLIMITED ON xstream_tbs;


GRANT CREATE SESSION TO xstrm;
GRANT SELECT ON V_$DATABASE to xstrm;
GRANT FLASHBACK ANY TABLE TO xstrm;


BEGIN
   DBMS_XSTREAM_AUTH.GRANT_ADMIN_PRIVILEGE(
      grantee                 => 'xstrmadmin',
      privilege_type          => 'CAPTURE',
      grant_select_privileges => TRUE
   );
END;

Create oracle outbound server with the tables you want to track.

DECLARE
  tables  DBMS_UTILITY.UNCL_ARRAY;
  schemas DBMS_UTILITY.UNCL_ARRAY;
BEGIN
  schemas(1) := 'my_schema';
  tables(1)  := 'ABC_CUSTOMER_PROPOSAL';
  tables(2)  := 'ABC_INPUT_PARSE';
  tables(3)  := 'ABC_PAR_VALUE';

  DBMS_XSTREAM_ADM.CREATE_OUTBOUND(
    server_name     =>  'dbzxout',
    table_names     =>  tables,
    schema_names    =>  schemas);
END;

Alright that is it for oracle part. Now lets download the required software.

Download Conflunet platform
Download Oracle Instant Client
Download JDK if you don't already have
Download Insomnia Rest Client and install it
Download Oracle JDBC Driver
Download following jar files from maven;
- ANTLR 4 Runtime
- Debezium Connector For Oracle
- Debezium Core
- Debezium ANTLR DDL Parsers

Now create a directory that will contain all the work we'll do here; BTW you can extract above downloaded files anywhere you like, but for simpliciy we will put them all under stream-demo folder.

mkdir stream-demo

Extract JDK under stream-demo/jdk
Extract Confuent to stream-demo/confluent
Extract Instant Clinet to stream-demo/instant-client
Create a directory stream-demo/confluent/share/java/kafka-connect-debezium
Place the following jars under stream-demo/confluent/share/java/kafka-connect-debezium
- 4 Runtime
- Debezium Connector For Oracle
- Debezium Core
- Debezium ANTLR DDL Parsers
- Oracle JDBC Driver
Also copy xstreams.jar from stream-demo/instant-client to stream-demo/confluent/share/java/kafka-connect-debezium
Add LD_LIBRARY_PATH and JAVA_HOME to your .profile like this;

export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/full/path/to/stream-demo/instant-client
export JAVA_HOME=/full/path/to/stream-demo/jdk

Save and quit then execute; source ~/.profile
Go to stream-demo/confluent/bin and execute ./confluent start

Ok we are almost done;
Open Insomnia rest client and create a work-space.
Add a new post request with the following payload

{
 "name": "gds",
 "config": {
  "connector.class" : "io.debezium.connector.oracle.OracleConnector",
  "tasks.max" : "1",
  "database.server.name" : "ORCL",
  "database.hostname" : "localhost",
  "database.port" : "1530",
  "database.user" : "xstrmadm",
  "database.password" : "oracle",
  "database.dbname" : "ORCL",
  "database.tablename.case.insensitive": false,
  "database.oracle.version": "11", "table.whitelist":"ORCL.MY_SCHEMA.ABC_INPUT_PARSE,ORCL.MY_SCHEMA.ABC_PAR_VALUE,ORCL.MY_SCHEMA.ABC_CUSTOMER_PROPOSAL",
  "database.out.server.name" : "dbzxout",
  "database.history.kafka.bootstrap.servers" : "localhost:9092",
  "database.history.kafka.topic": "schema-changes.inventory",
  "decimal.handling.mode": "precise"
 }
}

Change the above parameters according to your installation.

The URL for the post request is;
http://localhost:8083/connectors

Send the post request.
Send a get request to http://localhost:8083/connectors/gds/status

You should see something like;

{
  "name": "gds",
  "connector": {
    "state": "RUNNING",
    "worker_id": "127.0.1.1:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "127.0.1.1:8083"
    }
  ],
  "type": "source"
}

Ok now open ksql client;

cd stream-demo/confluent/bin
./ksql

If your tables are receiving some some dml already, you should see the topics by executing;

LIST TOPICS;

If you do not see anything try to insert, update or delete some records to your tables and then exec LIST TOPICS; again.

We have integrated oracle and kafka. Rest of the work is relatively easier. If we look our example architecture there is a python layer in which we read kafka topics with confluent python client and create streams from the topics; An example stream is created like this;

import logging

from streams.ksql_client import KSQLClient
from admin import ConfluentAdmin

class CustomerMaxSysId(KSQLClient, ConfluentAdmin):
    def __init__(self, *args, **kwargs):
        self.logger = logging.getLogger('streams.gr_large_none')
        super().__init__(*args, **kwargs)

    def create(self):
        name = 'S_CUSTOMER_MAX_SYS_ID'
        self.logger.info(f"Creating stream {name}")

        if self.is_stream_exists(name):
            self.logger.info(f"Stream already exists: {name}")
            return

        q = f"""
            create stream {name} (
                process_time    bigint,
                customer_no     varchar,
                sys_id          varchar,
                proposal_no     varchar,
                customer_type   varchar,
                flow_id         integer
            )
            WITH (kafka_topic='CUSTOMER_MAX_SYS_ID', value_format='json', timestamp='process_time')
        """
        self.ksql_client.ksql(q)
        self.logger.info(f"Created stream {name}")

We also use python ksql client to interact with ksql-rest api. You can also use rest api directly but using the ksql client api makes it more expressive.

Without getting into full detail in this layer we create streams both from kafka topics and by joining streams. An example stream-stream join can be like this;

import logging

from streams.ksql_client import KSQLClient
from admin import ConfluentAdmin

class GeneralLimitOutZero(KSQLClient, ConfluentAdmin):
    def __init__(self, *args, **kwargs):
        self.logger = logging.getLogger('streams.gr_large_none')
        super().__init__(*args, **kwargs)

    def create(self):
        name = 'S_GENERAL_LIMIT_OUT_ZERO'
        self.logger.info(f"Creating stream {name}")

        if self.is_stream_exists(name):
            self.logger.info(f"Stream already exists: {name}")
            return

        q = f"""
            create stream {name}
            WITH (kafka_topic='{name}', value_format='JSON')
            as
            select
                s.customer_no,
                s.sys_id,
                s.proposal_id,
                o.out_value
            from
                S_GR_LARGE_MODEL_ACTION s
            inner join S_OUTPUT_PARSE o within 15 days
                on s.sys_id = o.ref_no
            where
                s.out_value  = '0' and
                o.out_param  = 'LIMIT'
        """

        self.ksql_client.ksql(q)
        self.logger.info(f"Created stream {name}")

This layer is like an ETL process in a standard DWH except we use data streams instead of static data.

Next after creating streams, we can listen for data and take necessary actions or just report them in a reporting layer.

In our scenario we store the received data in redis. We use redis python client. And also we publish our received events to a redis channel.

        key = f"gds:alert:gr_large_none:{customer_no}:{sys_id}:{proposal_id}"
        self.__set_ttl(key)
        self.logger.debug(f'Setting key {key}')
        self.connection.hmset(key, record)
        channel = 'channel:gds:alert:gr_large_none'
        self.logger.debug(f'Publishing key to {channel}')
        self.connection.publish(channel, key)

You can publish events to a kafka topic instead of redis channel. We have chosen redis channel here so the next layers only interacts with redis instead of both redis and kafka.

The next and the final part is listening redis channels and producing actions; In this layer we have used go and before you ask there is no special purpose of choosing go over python, we have just wanted to demonstrate a heterogeneous, flexible environment with different technologies.

func Run() {

    logrus.Debug("Attaching redis channels ...")
    defer client.Close()
    pubsub := client.Subscribe(channelNames()...)
    defer pubsub.Close()
    _, err := pubsub.Receive()
    if err != nil {
        panic(err)
    }
    ch := pubsub.Channel()

    var anomalies = collectAnomalies()

    for channel, keys := range anomalies {
        if len(keys) >= anomalyLimit(channel) {
            notify(channel, keys)
            anomalies[channel] = []string{}
        }
    }

    for message := range ch {
        ca := channelAlias(message.Channel)
        key := message.Payload
        logrus.Debug(ca, " : Received message ", key)
        keys, ok := anomalies[message.Payload]
        if !ok {
            keys = []string{key}
            anomalies[message.Payload] = keys
        }

        anomalies[message.Channel] = util.AppendUniqueStr(anomalies[message.Channel], key)
        logrus.Debug(ca, " : Number of anomalies ", len(anomalies[message.Channel]))
        if len(anomalies[message.Channel]) >= anomalyLimit(message.Channel) {
            notify(message.Channel, anomalies[message.Channel])
            anomalies[message.Channel] = []string{}
        }
    }
}

Conclusion

The basic part of this scenario is actually integrating oracle and kafka using debezium. The later steps can be implemented in many different ways depending the case.

Thanks for reading.