** mode('streaming')*CREATE OR REPLACE TABLE eoj_table (`pk` string,`id` string,`name` str