{"rsdb":{"rid":"309138","subhead":"","postdate":"0","aid":"224300","fid":"116","uid":"1","topic":"1","content":"<div id=\"article_content\" class=\"article_content clearfix csdn-tracking-statistics\" data-pid=\"blog\" data-mod=\"popu_307\" data-dsm=\"post\">\n            <link rel=\"stylesheet\" href=\"https:\/\/csdnimg.cn\/release\/phoenix\/template\/css\/ck_htmledit_views-f57960eb32.css\">\n                              <link rel=\"stylesheet\" href=\"https:\/\/csdnimg.cn\/release\/phoenix\/template\/css\/ck_htmledit_views-f57960eb32.css\">\n          <div class=\"htmledit_views\" id=\"content_views\">\n            <div class=\"content-detail markdown-body\">\n                    \n<h2>1.\u524d\u8a00<\/h2>\n<p>E-MapReduce\u8ba1\u5212\u4eceEMR-3.18.1\u7248\u672c\u5f00\u59cb\u63d0\u4f9bSpark Streaming SQL\u7684\u9884\u89c8\u7248\u529f\u80fd\u3002Spark Streaming SQL\u662f\u5728Spark Structured Streaming\u7684\u57fa\u7840\u4e0a\u505a\u4e86\u8fdb\u4e00\u6b65\u5c01\u88c5\uff0c\u65b9\u4fbf\u7528\u6237\u4f7f\u7528SQL\u8bed\u8a00\u8fdb\u884cSpark\u6d41\u5f0f\u5206\u6790\u5f00\u53d1\u3002Spark Streaming SQL\u76f4\u63a5\u5730\u900f\u660e\u5730\u53d7\u60e0\u4e8eSpark SQL\u7684\u4f18\u5316\u5e26\u6765\u7684\u6027\u80fd\u63d0\u5347\uff0c\u540c\u65f6\u4e5f\u9075\u5faaSpark Structured Streaming\u7684\u8bed\u6cd5\u7ea6\u675f\uff0c\u4f8b\u5982Spark Structured Streaming\u4e0d\u652f\u6301\u591a\u6d41\u805a\u5408\u67e5\u8be2\uff0cSpark Streaming SQL\u4e5f\u5c31\u540c\u6837\u4e0d\u652f\u6301\u3002\u5173\u4e8eEMR Spark Streaming SQL\u7684\u4f7f\u7528\u5165\u95e8\uff0c\u8bf7\u53c2\u8003\uff1a<\/p>\n<ul><li><a href=\"https:\/\/yuque.antfin-inc.com\/emr\/emr-stream-compute\/dxb6hm\" rel=\"nofollow\">EMR Spark Streaming SQL\u8bed\u6cd5\u53c2\u8003<\/a><\/li>\n<li>\n<a href=\"https:\/\/github.com\/aliyun\/aliyun-emapreduce-sdk\/tree\/master-2.x\/emr-baseline-testing\" rel=\"nofollow\">Spark Streaming SQL Baseline Testing<\/a>\uff0c\u9a8c\u8bc1\u548c\u6d4b\u8bd5\u5de5\u5177\uff0c\u53ef\u4ee5\u4f5c\u4e3a\u5f00\u53d1\u5165\u95e8\u4f7f\u7528<\/li>\n<\/ul><h2>2.\u6848\u4f8b\u4ecb\u7ecd<\/h2>\n<p>\u672c\u6587\u5c06\u4f7f\u7528Spark Streaming SQL\u5b9e\u73b0\u5bf9HDFS audilog\u7684\u5b9e\u65f6\u7edf\u8ba1\u5206\u6790\u3002\u6570\u636e\u6d41\u56fe\u5982\u4e0b\u6240\u793a\uff1a<br><img src=\"https:\/\/yqfile.alicdn.com\/db96967ea1f573eb3b4443c64090f024285f6fef.png\" alt=\"image\" title=\"image\"><\/p>\n<p>\u4e00\u5171\u5206\u4e3a4\u6b65\uff1a<\/p>\n<ul><li>\u91c7\u96c6HDFS\u7684audit-log\u5230LogService\u4e2d\u3002<\/li>\n<li>Spark Streaming SQL\u4eceLogService\u6d88\u8d39\u6570\u636e\u5e76\u8fdb\u884c\u5206\u6790\u3002<\/li>\n<li>\u5206\u6790\u7ed3\u679c\u5199\u5230\u5230\u5b58\u50a8\u7cfb\u7edf\u4e2d\u3002<\/li>\n<li>\u4e0b\u6e38\u7cfb\u7edf\u4ece\u5b58\u50a8\u7cfb\u7edf\u4e2d\u67e5\u8be2\u548c\u4f7f\u7528\u5206\u6790\u7ed3\u679c\uff0c\u5178\u578b\u7684\u4e0b\u6e38\u7cfb\u7edf\u6709\u62a5\u8868\u6216\u8005\u76d1\u63a7\u62a5\u8b66\u3002<\/li>\n<\/ul><p>\u672c\u6587\u4e13\u6ce8\u4e8e\u7b2c\u4e8c\u6b65\uff0c\u5373\u5982\u4f55\u4f7f\u7528Spark Streaming SQL\u5206\u6790LogService\u6570\u636e\u3002<\/p>\n<h2>3.\u6848\u4f8b\u5b9e\u73b0<\/h2>\n<p>\u672c\u6b21\u6f14\u793a\u4e2d\uff0c\u6211\u4eec\u5c06\u7ed3\u679c\u6570\u636e\u5199\u5230Kafka\u4e2d\u3002<\/p>\n<h3>3.1 \u73af\u5883\u51c6\u5907<\/h3>\n<p>\u5728E-MapReduce\u4e2d\u521b\u5efa\u4e24\u4e2a\u96c6\u7fa4\uff1aHadoop\u96c6\u7fa4\u548cKafka\u96c6\u7fa4\u3002\u8fd9\u91cc\u5c31\u4e0d\u8be6\u7ec6\u6f14\u793a\u5982\u4f55\u521b\u5efaEMR\u96c6\u7fa4\u4e86\uff0c\u4e0d\u719f\u6089\u7684\u53ef\u4ee5\u53c2\u8003<a href=\"https:\/\/help.aliyun.com\/document_detail\/28088.html\" rel=\"nofollow\">\u5e2e\u52a9\u6587\u6863<\/a>\u3002<\/p>\n<h3>3.2 \u521b\u5efa\u6570\u636e\u8868<\/h3>\n<ul><li>\u521b\u5efaloghub\u6570\u636e\u6e90\u8868<\/li><\/ul><pre><code>CREATE DATABASE IF NOT EXISTS default;\nUSE default;\nDROP TABLE IF EXISTS loghub_hdfs_auditlog;\n\nCREATE TABLE loghub_hdfs_auditlog\nUSING loghub\nOPTIONS (\nsls.project = \"${SLS_PROJECT_NAME}\",\nsls.store = \"${SLS_STORE_NAME}\",\naccess.key.id = \"${ACCESS_KEY_ID}\",\naccess.key.secret = \"${ACCESS_KEY_SECRET}\",\nendpoint = \"${SLS_ENDPOINT}\",\nzookeeper.connect.address = \"${ZOOKEEPER_ADDRESS}\");<\/code><\/pre>\n<ul><li>\u521b\u5efaKafka\u8f93\u51fa\u8868<\/li><\/ul><pre><code>CREATE DATABASE IF NOT EXISTS default;\nUSE default;\nDROP TABLE IF EXISTS kafka_hdfs_auditlog_analysis;\n\nCREATE TABLE kafka_hdfs_auditlog_analysis\nUSING kafka\nOPTIONS (\nkafka.bootstrap.servers = \"${BOOTSTRAP_SERVERS}\",\nsubscribe = 'temp_hdfs_auditlog_analysis',\noutput.mode = 'complete',\nkafka.schema.registry.url = \"${SCHEMA_REGISTRY_URL}\",\nkafka.schema.record.name = \"${SCHEMA_RECORD_NAME}\",\nkafka.schema.record.namespace = \"${SCHEMA_RECORD_NAMESPACE}\");<\/code><\/pre>\n<p><strong>\u6ce8\u610f\uff1a<\/strong><\/p>\n<ol><li><strong>\u5177\u4f53\u505a\u67e5\u8be2\u5206\u6790\u65f6\uff0c\u9700\u8981\u5355\u72ec\u521b\u5efa\u4e00\u4e2aKafka\u7ed3\u679c\u8868\u3002\u8fd9\u662f\u56e0\u4e3a\u6bcf\u4e2a\u67e5\u8be2\u7ed3\u679c\u4e0d\u4e00\u6837\uff0c\u8f93\u51fa\u8868\u7684schema\u4e5f\u5c31\u4e0d\u4e00\u6837\u3002<\/strong><\/li>\n<li><strong>\u6839\u636e\u5177\u4f53\u7684\u67e5\u8be2\u7ed3\u679cschema\uff0c\u6211\u4eec\u9700\u8981\u9884\u5148\u5728Kafka Schema Registry\u4e2d\u6ce8\u518c\u597dKafka Topic Schema\u4fe1\u606f\u3002<\/strong><\/li>\n<\/ol><ul><li>Schema\u5b9a\u4e49\u793a\u4f8b\uff1a<\/li><\/ul><pre><code>{\"namespace\": \"org.apache.spark.emr.baseline.testing\",\n \"type\": \"record\",\n \"name\": \"TempResult\",\n \"fields\": [\n    {\"name\": \"avg_ss_quantity\", \"type\": [\"double\", \"null\"]},\n    {\"name\": \"avg_ss_ext_sales_price\", \"type\": [{\"type\": \"bytes\", \"logicalType\": \"decimal\", \"precision\": 11, \"scale\": 6}, \"null\"]},\n    {\"name\": \"avg_ss_ext_wholesale_cost\", \"type\": [{\"type\": \"bytes\", \"logicalType\": \"decimal\", \"precision\": 11, \"scale\": 6}, \"null\"]},\n    {\"name\": \"sum_ss_ext_wholesale_cost\", \"type\": [{\"type\": \"bytes\", \"logicalType\": \"decimal\", \"precision\": 17, \"scale\": 2}, \"null\"]}\n ]\n}<\/code><\/pre>\n<p>\u5c06schema\u5b9a\u4e49\u4fdd\u5b58\u5230\u6587\u4ef6\u4e2d\uff0c\u5e76\u4f7f\u7528\u811a\u672c<a href=\"https:\/\/github.com\/aliyun\/aliyun-emapreduce-sdk\/blob\/master-2.x\/emr-baseline-testing\/bin\/schema_register.py\" rel=\"nofollow\">\u5de5\u5177<\/a>\u6ce8\u518cschema\u5230Kafka Schema Registry\u4e2d\u3002<\/p>\n<pre><code><a href=https:\/\/www.cppentry.com style=text-decoration:underline;font-size:14px;color:#F70968; target=_blank>python<\/a> .\/schema_register.py ${SCHEMA_REGISTRY_URL} ${TOPIC_NAME} ${SCHEMA_FILE}<\/code><\/pre>\n<h3>3.3 \u7edf\u8ba15\u5206\u949f\u5185\u5404\u4e2a\u64cd\u4f5c\u6b21\u6570<\/h3>\n<ul><li>\u6ce8\u518cKafka Topic Schema\uff1a<\/li><\/ul><pre><code>{\"type\":\"record\",\n \"name\":\"TempResult\",\n \"namespace\":\"org.apache.spark.sql.streaming.test\",\n \"fields\":[\n    {\"name\":\"op\",\"type\":[\"string\",\"null\"]},\n    {\"name\":\"count\",\"type\":\"long\"},\n    {\"name\":\"window_time\",\"type\":\n        {\"type\":\"record\",\n         \"name\":\"window_time\",\n         \"namespace\":\"org.apache.spark.sql.streaming.test.window_time\",\n         \"fields\":[\n            {\"name\":\"start\",\"type\":[{\"type\":\"long\",\"logicalType\":\"timestamp-millis\"},\"null\"]},\n            {\"name\":\"end\",\"type\":[{\"type\":\"long\",\"logicalType\":\"timestamp-millis\"},\"null\"]}\n         ]\n        }\n    }\n  ]\n}<\/code><\/pre>\n<ul><li>\u521b\u5efa\u4e00\u4e2a\u7ed3\u679c\u8868\uff0c\u547d\u540d\u4e3a\uff1a\u201ckafka_hdfs_auditlog_op_count_in_5_mins\u201d\u3002<\/li><\/ul><pre><code>DROP TABLE IF EXISTS kafka_hdfs_auditlog_op_count_in_5_mins;\n\nCREATE TABLE kafka_hdfs_auditlog_op_count_in_5_mins\nUSING kafka\nOPTIONS (\nkafka.bootstrap.servers = \"${BOOTSTRAP_SERVERS}\",\nsubscribe = \"kafka_hdfs_auditlog_op_count_in_5_mins\",\noutput.mode = \"append\",\nkafka.schema.registry.url = \"${SCHEMA_REGISTRY_URL}\",\nkafka.schema.record.name = \"TempResult\",\nkafka.schema.record.namespace = \"org.apache.spark.sql.streaming.test\");<\/code><\/pre>\n<ul><li>\u67e5\u8be2\u5206\u6790<\/li><\/ul><pre><code>SET streaming.query.name=hdfs_auditlog_op_count_in_5_mins;\nSET spark.sql.streaming.checkpointLocation.hdfs_auditlog_op_count_in_5_mins=\/tmp\/spark\/sql\/streaming\/hdfs_auditlog_op_count_in_5_mins;\n\nINSERT INTO kafka_hdfs_auditlog_op_count_in_5_mins\nSELECT cmd op, count(*) count, window window_time\nFROM loghub_hdfs_auditlog\nGROUP BY TUMBLING(__time__, interval 5 minute), op\nHAVING delay(__time__) &lt; '5 minutes';<\/code><\/pre>\n<p><strong>\u6ce8\uff1a<\/strong><\/p>\n<ol><li><strong>\u201cstreaming.query.name\u201d\u53ef\u4ee5\u4e3a\u4efb\u610f\u540d\u5b57\u3002\u5fc5\u586b\u914d\u7f6e\u3002<\/strong><\/li>\n<li><strong>\u201cspark.sql.streaming.checkpointLocation.${streaming.query.name}\u201d\u5bf9\u201cstreaming.query.name\u201d\u7684job\u914d\u7f6echeckpoint\u8def\u5f84\uff0c\u53ef\u4ee5\u4e3a\u4efb\u610fhdfs\u8def\u5f84\u3002\u5fc5\u586b\u914d\u7f6e\u3002<\/strong><\/li>\n<\/ol><ul><li>\u67e5\u770b\u7ed3\u679c<\/li><\/ul><pre><code>\u901a\u8fc7Kafka\u547d\u4ee4\u884c\uff0c\u67e5\u770bkafka_hdfs_auditlog_op_count_in_5_mins topic\u6570\u636e\uff1a\n\n\u7a97\u53e3: 1550493600000 ~ 1550493900000\n{\"op\":{\"string\":\"create\"},\"count\":47438,\"window_time\":{\"start\":{\"long\":1550493600000},\"end\":{\"long\":1550493900000}}}\n{\"op\":{\"string\":\"delete\"},\"count\":181197,\"window_time\":{\"start\":{\"long\":1550493600000},\"end\":{\"long\":1550493900000}}}\n{\"op\":{\"string\":\"getfileinfo\"},\"count\":265451,\"window_time\":{\"start\":{\"long\":1550493600000},\"end\":{\"long\":1550493900000}}}\n{\"op\":{\"string\":\"listStatus\"},\"count\":641205,\"window_time\":{\"start\":{\"long\":1550493600000},\"end\":{\"long\":1550493900000}}}\n{\"op\":{\"string\":\"mkdirs\"},\"count\":12171,\"window_time\":{\"start\":{\"long\":1550493600000},\"end\":{\"long\":1550493900000}}}\n{\"op\":{\"string\":\"append\"},\"count\":30981,\"window_time\":{\"start\":{\"long\":1550493600000},\"end\":{\"long\":1550493900000}}}\n{\"op\":{\"string\":\"rename\"},\"count\":169709,\"window_time\":{\"start\":{\"long\":1550493600000},\"end\":{\"long\":1550493900000}}}\n{\"op\":{\"string\":\"setPermission\"},\"count\":52164,\"window_time\":{\"start\":{\"long\":1550493600000},\"end\":{\"long\":1550493900000}}}\n{\"op\":{\"string\":\"open\"},\"count\":436877,\"window_time\":{\"start\":{\"long\":1550493600000},\"end\":{\"long\":1550493900000}}}<\/code><\/pre>\n<h3>3.4 \u7edf\u8ba11\u5206\u949f\u5185\u5404\u4e2a\u6765\u6e90IP\u7684open\u64cd\u4f5c\u6b21\u6570<\/h3>\n<ul><li>\u6ce8\u518cKafka Topic Schema\uff1a<\/li><\/ul><pre><code>{\"type\":\"record\",\n \"name\":\"TempResult\",\n \"namespace\":\"org.apache.spark.sql.streaming.test\",\n \"fields\":[\n    {\"name\":\"ip\",\"type\":[\"string\",\"null\"]},\n    {\"name\":\"count\",\"type\":\"long\"},\n    {\"name\":\"window_time\",\"type\":\n        {\"type\":\"record\",\n         \"name\":\"window_time\",\n         \"namespace\":\"org.apache.spark.sql.streaming.test.window_time\",\n         \"fields\":[\n            {\"name\":\"start\",\"type\":[{\"type\":\"long\",\"logicalType\":\"timestamp-millis\"},\"null\"]},\n            {\"name\":\"end\",\"type\":[{\"type\":\"long\",\"logicalType\":\"timestamp-millis\"},\"null\"]}\n         ]\n        }\n    }\n  ]\n}<\/code><\/pre>\n<ul><li>\u521b\u5efa\u4e00\u4e2a\u7ed3\u679c\u8868\uff0c\u547d\u540d\u4e3a\uff1a\u201ckafka_hdfs_auditlog_read_count_per_ip_in_1_mins\u201d\u3002<\/li><\/ul><pre><code>DROP TABLE IF EXISTS kafka_hdfs_auditlog_read_count_per_ip_in_1_mins;\n\nCREATE TABLE kafka_hdfs_auditlog_read_count_per_ip_in_1_mins\nUSING kafka\nOPTIONS (\nkafka.bootstrap.servers = \"${BOOTSTRAP_SERVERS}\",\nsubscribe = \"kafka_hdfs_auditlog_read_count_per_ip_in_1_mins\",\noutput.mode = \"append\",\nkafka.schema.registry.url = \"${SCHEMA_REGISTRY_URL}\",\nkafka.schema.record.name = \"TempResult\",\nkafka.schema.record.namespace = \"org.apache.spark.sql.streaming.test\");<\/code><\/pre>\n<ul><li>\u67e5\u8be2\u5206\u6790<\/li><\/ul><pre><code>SET streaming.query.name=hdfs_auditlog_read_count_per_ip_in_1_mins;\nSET spark.sql.streaming.checkpointLocation.hdfs_auditlog_read_count_per_ip_in_1_mins=\/tmp\/spark\/sql\/streaming\/hdfs_auditlog_read_count_per_ip_in_1_mins;\n\nINSERT INTO kafka_hdfs_auditlog_read_count_per_ip_in_1_mins\nSELECT ip, count(*) count, window window_time\nFROM loghub_hdfs_auditlog\nwhere cmd='open'\nGROUP BY TUMBLING(__time__, interval 1 minute), ip\nHAVING delay(__time__) &lt; '1 minutes';<\/code><\/pre>\n<ul><li>\u67e5\u770b\u7ed3\u679c<\/li><\/ul><pre><code>\u7a97\u53e3: 1550540940000 ~ 1550541000000\n{\"ip\":{\"string\":\"172.*.*.130\"},\"count\":69090,\"window_time\":{\"start\":{\"long\":1550540940000},\"end\":{\"long\":1550541000000}}}\n{\"ip\":{\"string\":\"172.*.*.1\"},\"count\":69266,\"window_time\":{\"start\":{\"long\":1550540940000},\"end\":{\"long\":1550541000000}}}\n{\"ip\":{\"string\":\"10.*.*.1\"},\"count\":5129,\"window_time\":{\"start\":{\"long\":1550540940000},\"end\":{\"long\":1550541000000}}}\n{\"ip\":{\"string\":\"172.*.*.52\"},\"count\":70469,\"window_time\":{\"start\":{\"long\":1550540940000},\"end\":{\"long\":1550541000000}}}\n{\"ip\":{\"string\":\"10.*.*.24\"},\"count\":7206,\"window_time\":{\"start\":{\"long\":1550540940000},\"end\":{\"long\":1550541000000}}}\n{\"ip\":{\"string\":\"172.*.*.48\"},\"count\":136101,\"window_time\":{\"start\":{\"long\":1550540940000},\"end\":{\"long\":1550541000000}}}\n{\"ip\":{\"string\":\"172.*.*.19\"},\"count\":67226,\"window_time\":{\"start\":{\"long\":1550540940000},\"end\":{\"long\":1550541000000}}}\n{\"ip\":{\"string\":\"172.*.*.82\"},\"count\":141886,\"window_time\":{\"start\":{\"long\":1550540940000},\"end\":{\"long\":1550541000000}}}\n{\"ip\":{\"string\":\"172.*.*.50\"},\"count\":69165,\"window_time\":{\"start\":{\"long\":1550540940000},\"end\":{\"long\":1550541000000}}}\n{\"ip\":{\"string\":\"172.*.*.54\"},\"count\":151539,\"window_time\":{\"start\":{\"long\":1550540940000},\"end\":{\"long\":1550541000000}}}\n\n\u7a97\u53e3: 1550541000000 ~ 1550541060000\n{\"ip\":{\"string\":\"192.*.*.22\"},\"count\":77776,\"window_time\":{\"start\":{\"long\":1550541000000},\"end\":{\"long\":1550541060000}}}\n{\"ip\":{\"string\":\"10.*.*.111\"},\"count\":9373,\"window_time\":{\"start\":{\"long\":1550541000000},\"end\":{\"long\":1550541060000}}}\n{\"ip\":{\"string\":\"10.*.*.2\"},\"count\":9329,\"window_time\":{\"start\":{\"long\":1550541000000},\"end\":{\"long\":1550541060000}}}\n\n\u7a97\u53e3: 1550541060000 ~ 1550541120000\n{\"ip\":{\"string\":\"172.*.*.207\"},\"count\":10481,\"window_time\":{\"start\":{\"long\":1550541060000},\"end\":{\"long\":1550541120000}}}\n{\"ip\":{\"string\":\"192.*.*.138\"},\"count\":28965,\"window_time\":{\"start\":{\"long\":1550541060000},\"end\":{\"long\":1550541120000}}}\n{\"ip\":{\"string\":\"10.*.*.160\"},\"count\":22015,\"window_time\":{\"start\":{\"long\":1550541060000},\"end\":{\"long\":1550541120000}}}\n{\"ip\":{\"string\":\"172.*.*.234\"},\"count\":32892,\"window_time\":{\"start\":{\"long\":1550541060000},\"end\":{\"long\":1550541120000}}}<\/code><\/pre>\n<h3>3.5 \u7edf\u8ba11\u5206\u949f\u5185QPS\u8d85\u8fc7100\u7684\u6765\u6e90IP<\/h3>\n<ul><li>\u6ce8\u518cKafka Topic Schema\uff1a<\/li><\/ul><pre><code>{\"type\":\"record\",\n \"name\":\"TempResult\",\n \"namespace\":\"org.apache.spark.sql.streaming.test\",\n \"fields\":[\n    {\"name\":\"ip\",\"type\":[\"string\",\"null\"]},\n    {\"name\":\"qps\",\"type\":[\"double\",\"null\"]},\n    {\"name\":\"window_time\",\"type\":\n        {\"type\":\"record\",\n         \"name\":\"window_time\",\n         \"namespace\":\"org.apache.spark.sql.streaming.test.window_time\",\n         \"fields\":[\n            {\"name\":\"start\",\"type\":[{\"type\":\"long\",\"logicalType\":\"timestamp-millis\"},\"null\"]},\n            {\"name\":\"end\",\"type\":[{\"type\":\"long\",\"logicalType\":\"timestamp-millis\"},\"null\"]}\n         ]\n        }\n    }\n  ]\n}<\/code><\/pre>\n<ul><li>\u521b\u5efa\u4e00\u4e2a\u7ed3\u679c\u8868\uff0c\u547d\u540d\u4e3a\uff1a\u201ckafka_hdfs_auditlog_qps_gt_100_ip_in_1_mins\u201d\u3002<\/li><\/ul><pre><code>DROP TABLE IF EXISTS kafka_hdfs_auditlog_qps_gt_100_ip_in_1_mins;\n\nCREATE TABLE kafka_hdfs_auditlog_qps_gt_100_ip_in_1_mins\nUSING kafka\nOPTIONS (\nkafka.bootstrap.servers = \"${BOOTSTRAP_SERVERS}\",\nsubscribe = \"kafka_hdfs_auditlog_qps_gt_100_ip_in_1_mins\",\noutput.mode = \"append\",\nkafka.schema.registry.url = \"${SCHEMA_REGISTRY_URL}\",\nkafka.schema.record.name = \"TempResult\",\nkafka.schema.record.namespace = \"org.apache.spark.sql.streaming.test\");<\/code><\/pre>\n<ul><li>\u67e5\u8be2\u5206\u6790<\/li><\/ul><pre><code>SET streaming.query.name=hdfs_auditlog_qps_gt_100_ip_in_1_mins;\nSET spark.sql.streaming.checkpointLocation.hdfs_auditlog_qps_gt_100_ip_in_1_mins=\/tmp\/spark\/sql\/streaming\/hdfs_auditlog_qps_gt_100_ip_in_1_mins;\n\nINSERT INTO kafka_hdfs_auditlog_qps_gt_100_ip_in_1_mins\nSELECT ip, qps, window window_time\nFROM \n  (SELECT ip, count(cmd)\/60 qps, window\n  FROM loghub_hdfs_auditlog\n  GROUP BY TUMBLING(__time__, interval 1 minute), ip\n  HAVING delay(__time__) &lt; '1 minutes') t\nWHERE qps &gt; 100;<\/code><\/pre>\n<ul><li>\u67e5\u770b\u7ed3\u679c<\/li><\/ul><pre><code>\u7a97\u53e3\uff1a1550543160000 ~ 1550543220000\n{\"ip\":{\"string\":\"10.*.*.140\"},\"qps\":{\"double\":328.6333333333333},\"window_time\":{\"start\":{\"long\":1550543160000},\"end\":{\"long\":1550543220000}}}\n{\"ip\":{\"string\":\"172.*.*.13\"},\"qps\":{\"double\":276.43333333333334},\"window_time\":{\"start\":{\"long\":1550543160000},\"end\":{\"long\":1550543220000}}}\n{\"ip\":{\"string\":\"172.*.*.27\"},\"qps\":{\"double\":228.23333333333332},\"window_time\":{\"start\":{\"long\":1550543160000},\"end\":{\"long\":1550543220000}}}\n{\"ip\":{\"string\":\"192.*.*.170\"},\"qps\":{\"double\":407.4},\"window_time\":{\"start\":{\"long\":1550543160000},\"end\":{\"long\":1550543220000}}}\n{\"ip\":{\"string\":\"172.*.*.95\"},\"qps\":{\"double\":233.6},\"window_time\":{\"start\":{\"long\":1550543160000},\"end\":{\"long\":1550543220000}}}\n{\"ip\":{\"string\":\"172.*.*.8\"},\"qps\":{\"double\":341.3833333333333},\"window_time\":{\"start\":{\"long\":1550543160000},\"end\":{\"long\":1550543220000}}}\n{\"ip\":{\"string\":\"192.*.*.167\"},\"qps\":{\"double\":357.73333333333335},\"window_time\":{\"start\":{\"long\":1550543160000},\"end\":{\"long\":1550543220000}}}\n{\"ip\":{\"string\":\"10.*.*.142\"},\"qps\":{\"double\":283.05},\"window_time\":{\"start\":{\"long\":1550543160000},\"end\":{\"long\":1550543220000}}}\n{\"ip\":{\"string\":\"192.*.*.98\"},\"qps\":{\"double\":371.46666666666664},\"window_time\":{\"start\":{\"long\":1550543160000},\"end\":{\"long\":1550543220000}}}\n{\"ip\":{\"string\":\"192.*.*.209\"},\"qps\":{\"double\":501.35},\"window_time\":{\"start\":{\"long\":1550543160000},\"end\":{\"long\":1550543220000}}}\n{\"ip\":{\"string\":\"10.*.*.12\"},\"qps\":{\"double\":276.8333333333333},\"window_time\":{\"start\":{\"long\":1550543160000},\"end\":{\"long\":1550543220000}}}\n{\"ip\":{\"string\":\"10.*.*.98\"},\"qps\":{\"double\":276.3833333333333},\"window_time\":{\"start\":{\"long\":1550543160000},\"end\":{\"long\":1550543220000}}}\n{\"ip\":{\"string\":\"172.*.*.169\"},\"qps\":{\"double\":231.86666666666667},\"window_time\":{\"start\":{\"long\":1550543160000},\"end\":{\"long\":1550543220000}}}\n{\"ip\":{\"string\":\"10.*.*.166\"},\"qps\":{\"double\":358.9},\"window_time\":{\"start\":{\"long\":1550543160000},\"end\":{\"long\":1550543220000}}}\n\n\u7a97\u53e3\uff1a1550543220000 ~ 1550543280000\n{\"ip\":{\"string\":\"172.*.*.41\"},\"qps\":{\"double\":675.3},\"window_time\":{\"start\":{\"long\":1550543220000},\"end\":{\"long\":1550543280000}}}\n{\"ip\":{\"string\":\"192.*.*.97\"},\"qps\":{\"double\":364.45},\"window_time\":{\"start\":{\"long\":1550543220000},\"end\":{\"long\":1550543280000}}}\n{\"ip\":{\"string\":\"172.*.*.17\"},\"qps\":{\"double\":392.98333333333335},\"window_time\":{\"start\":{\"long\":1550543220000},\"end\":{\"long\":1550543280000}}}\n{\"ip\":{\"string\":\"192.*.*.234\"},\"qps\":{\"double\":361.3833333333333},\"window_time\":{\"start\":{\"long\":1550543220000},\"end\":{\"long\":1550543280000}}}\n{\"ip\":{\"string\":\"192.*.*.72\"},\"qps\":{\"double\":354.98333333333335},\"window_time\":{\"start\":{\"long\":1550543220000},\"end\":{\"long\":1550543280000}}}\n{\"ip\":{\"string\":\"172.*.*.3\"},\"qps\":{\"double\":513.2833333333333},\"window_time\":{\"start\":{\"long\":1550543220000},\"end\":{\"long\":1550543280000}}}\n{\"ip\":{\"string\":\"10.*.*.1\"},\"qps\":{\"double\":435.18333333333334},\"window_time\":{\"start\":{\"long\":1550543220000},\"end\":{\"long\":1550543280000}}}\n{\"ip\":{\"string\":\"10.*.*.240\"},\"qps\":{\"double\":458.45},\"window_time\":{\"start\":{\"long\":1550543220000},\"end\":{\"long\":1550543280000}}}\n{\"ip\":{\"string\":\"10.*.*.85\"},\"qps\":{\"double\":500.46666666666664},\"window_time\":{\"start\":{\"long\":1550543220000},\"end\":{\"long\":1550543280000}}}\n{\"ip\":{\"string\":\"172.*.*.234\"},\"qps\":{\"double\":635.1333333333333},\"window_time\":{\"start\":{\"long\":1550543220000},\"end\":{\"long\":1550543280000}}}\n{\"ip\":{\"string\":\"172.*.*.107\"},\"qps\":{\"double\":371.76666666666665},\"window_time\":{\"start\":{\"long\":1550543220000},\"end\":{\"long\":1550543280000}}}\n{\"ip\":{\"string\":\"10.*.*.195\"},\"qps\":{\"double\":357.46666666666664},\"window_time\":{\"start\":{\"long\":1550543220000},\"end\":{\"long\":1550543280000}}}\n{\"ip\":{\"string\":\"10.*.*.31\"},\"qps\":{\"double\":178.95},\"window_time\":{\"start\":{\"long\":1550543220000},\"end\":{\"long\":1550543280000}}}<\/code><\/pre>\n<h2>4. \u5c0f\u7ed3<\/h2>\n<p>\u672c\u6587\u7b80\u8981\u793a\u4f8b\u4e86\u5982\u4f55\u5728EMR\u4e0a\u4f7f\u7528Spark Streaming SQL\u8fdb\u884c\u6d41\u5f0f\u5206\u6790\u3002\u9700\u8981\u6ce8\u610f\u7684\u662f\uff0cEMR Spark Streaming SQL\u5904\u4e8e\u9884\u89c8\u7248\u9636\u6bb5\uff0c\u8bed\u6cd5\u548c\u529f\u80fd\u8fd8\u5728\u4e0d\u65ad\u4e30\u5bcc\u5b8c\u5584\u4e2d\u3002<\/p>\n<ul><li>JIRA\uff1a<a href=\"https:\/\/issues.apache.org\/jira\/browse\/SPARK-24630\" rel=\"nofollow\">https:\/\/issues.apache.org\/jira\/browse\/SPARK-24630<\/a>\n<\/li>\n<li>Design doc\uff1a<a href=\"https:\/\/docs.google.com\/document\/d\/19degwnIIcuMSELv6BQ_1VQI5AIVcvGeqOm5xE2-aRA0\" rel=\"nofollow\">https:\/\/docs.google.com\/document\/d\/19degwnIIcuMSELv6BQ_1VQI5AIVcvGeqOm5xE2-aRA0<\/a>\n<\/li>\n<\/ul><\/div>          <\/div>\n                  <\/div>","orderid":"0","title":"\n &nbsp;  &nbsp;  &nbsp;  &nbsp;  &nbsp;  &nbsp;  &nbsp;  &nbsp;  &nbsp;  &nbsp;   \u57fa\u4e8eSpark SQL\u5b9e\u73b0\u5bf9HDFS\u64cd\u4f5c\u7684\u5b9e\u65f6\u76d1\u63a7\u62a5\u8b66\n &nbsp;  ","smalltitle":"","mid":"0","fname":"Spark","special_id":"0","bak_id":"0","info":"0","hits":"477","pages":"1","comments":"0","posttime":"2019-05-16 01:21:37","list":"1557940897","username":"admin","author":"","copyfrom":"","copyfromurl":"","titlecolor":"","fonttype":"0","titleicon":"0","picurl":"https:\/\/www.cppentry.com\/upload_files\/","ispic":"0","yz":"1","yzer":"","yztime":"0","levels":"0","levelstime":"0","keywords":"<A HREF='https:\/\/www.cppentry.com\/do\/search.php?type=keyword&keyword=%26nbsp' target=_blank>&nbsp<\/A> <A HREF='https:\/\/www.cppentry.com\/do\/search.php?type=keyword&keyword=%BB%F9%D3%DA' target=_blank>\u57fa\u4e8e<\/A> <A HREF='https:\/\/www.cppentry.com\/do\/search.php?type=keyword&keyword=Spark' target=_blank>Spark<\/A> <A HREF='https:\/\/www.cppentry.com\/do\/search.php?type=keyword&keyword=SQL' target=_blank>SQL<\/A> <A HREF='https:\/\/www.cppentry.com\/do\/search.php?type=keyword&keyword=%CA%B5%CF%D6' target=_blank>\u5b9e\u73b0<\/A> <A HREF='https:\/\/www.cppentry.com\/do\/search.php?type=keyword&keyword=HDFS' target=_blank>HDFS<\/A> <A HREF='https:\/\/www.cppentry.com\/do\/search.php?type=keyword&keyword=%B2%D9%D7%F7' target=_blank>\u64cd\u4f5c<\/A> <A HREF='https:\/\/www.cppentry.com\/do\/search.php?type=keyword&keyword=%CA%B5%CA%B1' target=_blank>\u5b9e\u65f6<\/A> <A HREF='https:\/\/www.cppentry.com\/do\/search.php?type=keyword&keyword=%BC%E0%BF%D8' target=_blank>\u76d1\u63a7<\/A> <A HREF='https:\/\/www.cppentry.com\/do\/search.php?type=keyword&keyword=%B1%A8%BE%AF' target=_blank>\u62a5\u8b66<\/A>","jumpurl":"","iframeurl":"","style":"","template":"a:3:{s:4:\"foot\";s:0:\"\";s:8:\"bencandy\";s:0:\"\";s:4:\"head\";s:0:\"\";}","target":"0","ip":"47.106.78.186","lastfid":"0","money":"0","buyuser":"","passwd":"","allowdown":"","allowview":"","editer":"","edittime":"0","begintime":"0","endtime":"0","description":"                                                                                    1.\u524d\u8a00E-MapReduce\u8ba1\u5212\u4eceEMR-3.18.1\u7248\u672c\u5f00\u59cb\u63d0\u4f9bSpark Streaming SQL\u7684\u9884\u89c8\u7248\u529f\u80fd\u3002Spark Streaming SQL\u662f\u5728Spark Structured Streaming\u7684\u57fa\u7840\u4e0a\u505a\u4e86\u8fdb\u4e00\u6b65\u5c01\u88c5\uff0c\u65b9\u4fbf\u7528\u6237\u4f7f\u7528..","lastview":"1713504451","digg_num":"0","digg_time":"0","forbidcomment":"0","ifvote":"0","heart":"","htmlname":"","city_id":"0"},"page":"1"}