Guadazi
0%

Flink AsyncApiScalaSink

Posted on Edited on In

Flink AsyncApiScalaSink

andrew建议：

  • onCompeleted没有把IOException暴露给用户
  • 链接配置之类的参数如何传递进去
  • 反压机制如何做？sink经常遇到的问题，下游抖动会直接导致数据丢失
  • 反压 限流 重试
  • 支持batch，条数控制
  • checkpoint时，需要batch立即flush出去

Http Client

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
<dependency>
    <groupId>org.apache.httpcomponents</groupId>
    <artifactId>httpclient</artifactId>
    <version>4.5.2</version>
    <exclusions>
        <exclusion>
            <artifactId>commons-logging</artifactId>
            <groupId>commons-logging</groupId>
        </exclusion>
    </exclusions>
</dependency>

<dependency>
    <groupId>org.apache.httpcomponents</groupId>
    <artifactId>httpcore</artifactId>
    <version>4.4.5</version>
</dependency>

<dependency>
    <groupId>org.apache.httpcomponents</groupId>
    <artifactId>httpcore-nio</artifactId>
    <version>4.4.5</version>
</dependency>

<dependency>
    <groupId>org.apache.httpcomponents</groupId>
    <artifactId>httpasyncclient</artifactId>
    <version>4.1.2</version>
    <exclusions>
        <exclusion>
            <artifactId>commons-logging</artifactId>
            <groupId>commons-logging</groupId>
        </exclusion>
    </exclusions>
</dependency>

AsyncApiScalaSink

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import java.io.IOException
import java.util

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.http._
import org.apache.http.client.entity.UrlEncodedFormEntity
import org.apache.http.client.methods.{HttpEntityEnclosingRequestBase, HttpPost}
import org.apache.http.concurrent.FutureCallback
import org.apache.http.impl.client.DefaultConnectionKeepAliveStrategy
import org.apache.http.impl.nio.client.{CloseableHttpAsyncClient, HttpAsyncClients}
import org.apache.http.message.BasicNameValuePair
import org.apache.http.util.EntityUtils

class AsyncApiScalaSink[E](httpInvoker: HttpInvoker[E]) extends RichSinkFunction[E] {


  var httpClient: CloseableHttpAsyncClient = _

  override def open(parameters: Configuration): Unit = {
    httpClient = HttpAsyncClients.custom.setKeepAliveStrategy(DefaultConnectionKeepAliveStrategy.INSTANCE).build
    httpClient.start()
  }

  override def invoke(value: E, context: SinkFunction.Context[_]): Unit = {
    var httpEntity: HttpEntityEnclosingRequestBase = null
    val method: String = httpInvoker.getMethod
    val url: String = httpInvoker.getUrl
    if ("GET" == method) {
    }
    else if ("POST" == method) {
      val params1: util.Map[String, String] = httpInvoker.getParams(value)
      val params: util.List[NameValuePair] = new util.ArrayList[NameValuePair]
      if (params1 != null) {
        import scala.collection.JavaConversions._
        for (entry <- params1.entrySet) {
          val key: String = entry.getKey
          val value1: String = entry.getValue
          params.add(new BasicNameValuePair(key, value1))
        }
      }
      val entity: UrlEncodedFormEntity = new UrlEncodedFormEntity(params, Consts.UTF_8)
      httpEntity = new HttpPost(url)
      httpEntity.setEntity(entity)
    }

    httpClient.execute(httpEntity, new FutureCallback[HttpResponse]() {
      override def completed(response: HttpResponse): Unit = {
        val statusLine: StatusLine = response.getStatusLine
        val httpStatusCode: Int = statusLine.getStatusCode
        var content: String = null
        try {
          val entity: HttpEntity = response.getEntity
          content = EntityUtils.toString(entity)
        } catch {
          case e: IOException =>
            System.err.print(e.getMessage)
        }
        httpInvoker.onCompleted(value, httpStatusCode, content)
      }

      override

      def failed(ex: Exception): Unit = {
        httpInvoker.onFailed(value, ex)
      }

      override

      def cancelled(): Unit = {
        httpInvoker.onCanceled(value)
      }
    })
  }

  override def close(): Unit = {
    httpClient.close()
  }
}

HttpInvoker

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import java.io.Serializable;
import java.util.Map;

public interface HttpInvoker<E> extends Serializable {
    Map<String, String> getParams(E e);

    String getMethod();

    String getUrl();

    void onCompleted(E value, int httpStatusCode, String content);

    void onFailed(E value, Exception ex);

    void onCanceled(E value);
}