title: How to Integrate MySql tables Data To Ksql Stream or Tables?
body: <p>I am trying to build a data pipeline from <strong>MySql</strong> to <strong>Ksql</strong>.</p>
  
  <p>Use Case: data source is MySql. I have created a table in MySql.</p>
  
  <p>I am using </p>
  
  <pre><code>./bin/connect-standalone ./etc/schema-registry/connect-avro-standalone.properties  ./etc/kafka-connect-jdbc/source-quickstart-sqlite.properties 
  </code></pre>
  
  <p>to start a standalone connector. And it is working fine.</p>
  
  <p>I am starting the consumer with topic name i.e. </p>
  
  <pre><code>./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test1Category --from-beginning
  </code></pre>
  
  <p>when I am inserting the data in MySQL table I am getting the result in consumer as well. <strong>I have created KSQL Stream as will with the same topic name</strong>. I am expecting the same result in my Kstream as well, But i am not getting any result when i am doing </p>
  
  <pre><code>select * from &lt;streamName&gt;
  </code></pre>
  
  <p><strong>Connector configuration--source-quickstart-mysql.properties</strong></p>
  
  <pre><code>    name=jdbc_source_mysql
  connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
  
  key.converter=io.confluent.connect.avro.AvroConverter
  key.converter.schema.registry.url=http://localhost:8081
  
  value.converter=io.confluent.connect.avro.AvroConverter
  value.converter.schema.registry.url=http://localhost:8081
  
  connection.url=jdbc:mysql://localhost:3306/testDB?user=root&amp;password=cloudera
  
  #comment=Which table(s) to include
  table.whitelist=ftest
  
  mode=incrementing
  incrementing.column.name=id
  
  topic.prefix=ftopic
  </code></pre>
  
  <p><strong>Sample Data</strong></p>
  
  <ul>
  <li><strong>MySql</strong></li>
  </ul>
  
  <p>1.) Create Database: </p>
  
  <pre><code>CREATE DATABASE testDB;
  </code></pre>
  
  <p>2.) Use Database:</p>
  
  <pre><code>USE testDB;
  </code></pre>
  
  <p>3.) create the table:</p>
  
  <pre><code>    CREATE TABLE products (
    id INTEGER NOT NULL PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    description VARCHAR(512),
    weight FLOAT
  );
  </code></pre>
  
  <p>4.) Insert data into the table:</p>
  
  <pre><code>    INSERT INTO products(id,name,description,weight)
    VALUES (103,'car','Small car',20);
  </code></pre>
  
  <ul>
  <li><strong>KSQL</strong></li>
  </ul>
  
  <p>1.) Create Stream:</p>
  
  <pre><code>CREATE STREAM pro_original (id int, name varchar, description varchar,weight bigint) WITH \
  (kafka_topic='proproducts', value_format='DELIMITED');
  </code></pre>
  
  <p>2.) Select Query:</p>
  
  <pre><code>Select * from pro_original;
  </code></pre>
  
  <p><strong>Expected Output</strong></p>
  
  <ol>
  <li><strong>Consumer</strong></li>
  </ol>
  
  <p>getting the data which is inserted in the MySQL table.</p>
  
  <p>Here I am getting the data in MySQL.</p>
  
  <ol start="2">
  <li><strong>Ksql</strong></li>
  </ol>
  
  <p>In-Stream data should be populated which is inserted in Mysql table and reflecting in Kafka topic.</p>
  
  <p><strong>I am not getting expected result in ksql</strong></p>
  
  <p>Help me for this data pipeline.</p>