深度实践 | 自如基于Apache StreamPark 的实时计算平台实践

ldap:
  # Is ldap enabled?
  enable: false
  ## AD server IP, default port 389
  urls: ldap://99.99.99.99:389
  ## Login Account
  base-dn: dc=streampark,dc=com
  username: cn=Manager,dc=streampark,dc=com
  password: streampark
  user:
    identity-attribute: uid
    email-attribute: mail

SELECT * FROM orders;执行SQL

SELECT * FROM orders WHERE region = 'beijing';

SELECT * FROM orders WHERE region = 'hangzhou';

SELECT name, age, price, phone FROM user;

SELECT Encryption_function(name), age, price, Sensitive_field_functions(phone) FROM user;

SELECT name, Encryption_function(age), price, Sensitive_field_functions(phone) FROM user;

curl -X POST '/flink/app/start' \
-H 'Authorization: $token' \
-H 'Content-Type: application/x-www-form-urlencoded; charset=UTF-8' \
--data-urlencode 'savePoint=' \
--data-urlencode 'allowNonRestored=false' \
--data-urlencode 'savePointed=false' \
--data-urlencode 'id=100501'

FROM apache/flink:1.14.5-scala_2.11-java8
ENV TIME_ZONE=Asia/Shanghai
COPY ./conf /opt/hadoop/conf
COPY lib $FLINK_HOME/lib/

docker buildx build --push --platform linux/amd64 -t ${私有镜像仓库地址}

FROM apache/flink:1.14.5-scala_2.11-java8
ENV TIME_ZONE=Asia/Shanghai
COPY ./conf /opt/hadoop/conf
COPY lib $FLINK_HOME/lib/
RUN apt-get update --fix-missing && apt-get install -y fontconfig --fix-missing && \
    apt-get install -y openjdk-8-jdk && \
    apt-get install -y ant && \
    apt-get clean;

RUN apt-get install sudo -y

# Fix certificate issues
RUN apt-get update && \
    apt-get install ca-certificates-java && \
    apt-get clean && \
    update-ca-certificates -f;

# Setup JAVA_HOME -- useful for docker commandline
ENV JAVA_HOME /usr/lib/jvm/java-8-openjdk-amd64/
RUN export JAVA_HOME
RUN apt-get install -y unzip
RUN curl -Lo arthas-packaging-latest-bin.zip  'https://arthas.aliyun.com/download/latest_version?mirror=aliyun'
RUN unzip -d arthas-latest-bin arthas-packaging-latest-bin.zip

git clone https://github.com/apache/flink-shaded.git

cd /flink-1.14.5/conf
vim flink-conf.yaml

apiVersion: v1
kind: Pod
metadata:
  name: pod-template
spec:
  hostAliases:
    - ip: 10.216.xxx.79
      hostnames:
        - handoop1
    - hostnames:
        - handoop2
      ip: 10.16.xx.48
    - hostnames:
        - handoop3
      ip: 10.16.xx.49
    - hostnames:
        - handoop4
      ip: 10.16.xx.50
   .......

apiVersion: v1
data:
  resolv.conf: "nameserver  10.216.138.226" //DNS服务
kind: ConfigMap
metadata:
  creationTimestamp: "2022-07-13T10:16:18Z"
  managedFields:
  name: dns-configmap
  namespace: native-flink

export HADOOP_CONF_DIR=/home/streamx/conf

// 寻找hadoop配置文件的流程
//1、先去寻找是否添加了参数：kubernetes.hadoop.conf.config-map.name
@Override
public Optional<String> getExistingHadoopConfigurationConfigMap() {
    final String existingHadoopConfigMap =
            flinkConfig.getString(KubernetesConfigOptions.HADOOP_CONF_CONFIG_MAP);
    if (StringUtils.isBlank(existingHadoopConfigMap)) {
        return Optional.empty();
    } else {
        return Optional.of(existingHadoopConfigMap.trim());
    }
}

@Override
public Optional<String> getLocalHadoopConfigurationDirectory() {
    // 2、如果没有 1 中指定的参数，查找提交 native 命令的本地环境是否有环境变量：HADOOP_CONF_DIR
    final String hadoopConfDirEnv = System.getenv(Constants.ENV_HADOOP_CONF_DIR);
    if (StringUtils.isNotBlank(hadoopConfDirEnv)) {
        return Optional.of(hadoopConfDirEnv);
    }
    // 3、如果没有 2 中环境变量，再继续看否有环境变量：HADOOP_HOME
    final String hadoopHomeEnv = System.getenv(Constants.ENV_HADOOP_HOME);
    if (StringUtils.isNotBlank(hadoopHomeEnv)) {
        // Hadoop 2.2+
        final File hadoop2ConfDir = new File(hadoopHomeEnv, "/etc/hadoop");
        if (hadoop2ConfDir.exists()) {
            return Optional.of(hadoop2ConfDir.getAbsolutePath());
        }

        // Hadoop 1.x
        final File hadoop1ConfDir = new File(hadoopHomeEnv, "/conf");
        if (hadoop1ConfDir.exists()) {
            return Optional.of(hadoop1ConfDir.getAbsolutePath());
        }
    }

    return Optional.empty();
}

final List<File> hadoopConfigurationFileItems = getHadoopConfigurationFileItems(localHadoopConfigurationDirectory.get());
// 如果没有找到 1、2、3 说明没有 hadoop 环境
if (hadoopConfigurationFileItems.isEmpty()) {
    LOG.warn(
            "Found 0 files in directory {}, skip to mount the Hadoop Configuration ConfigMap.",
            localHadoopConfigurationDirectory.get());
    return flinkPod;
}
// 如果 2 或者 3 存在，会在路径下查找 core-site.xml 和 hdfs-site.xml 文件
private List<File> getHadoopConfigurationFileItems(String localHadoopConfigurationDirectory) {
    final List<String> expectedFileNames = new ArrayList<>();
    expectedFileNames.add("core-site.xml");
    expectedFileNames.add("hdfs-site.xml");

    final File directory = new File(localHadoopConfigurationDirectory);
    if (directory.exists() && directory.isDirectory()) {
        return Arrays.stream(directory.listFiles())
                .filter(
                        file ->
                                file.isFile()
                                        && expectedFileNames.stream()
                                                .anyMatch(name -> file.getName().equals(name)))
                .collect(Collectors.toList());
    } else {
        return Collections.emptyList();
    }
}

//如果有 hadoop 的环境，将会把上述两个文件解析为 kv 对，然后构建成一个 ConfigMap，名字命名规则如下
public static String getHadoopConfConfigMapName(String clusterId) {
    return Constants.HADOOP_CONF_CONFIG_MAP_PREFIX + clusterId;
}

netstat -tlnp | grep 10000
netstat -tlnp | grep 10002